288 lines
10 KiB
Go
288 lines
10 KiB
Go
package api
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"regexp"
|
|
"strconv"
|
|
"time"
|
|
|
|
"git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/insight"
|
|
)
|
|
|
|
var insightJobIDPattern = regexp.MustCompile(`^insightjob_[a-f0-9]{24}$`)
|
|
|
|
type insightService interface {
|
|
Latest(context.Context, int64, string, insight.Module) (insight.Snapshot, error)
|
|
Refresh(context.Context, int64, string, insight.Module, string) (insight.Job, bool, error)
|
|
RefreshSection(context.Context, int64, string, insight.Module, insight.Section, string) (insight.Job, bool, error)
|
|
Enqueue(context.Context, int64, string, insight.Module, string) (insight.Job, bool, error)
|
|
GetJob(context.Context, int64, string) (insight.Job, error)
|
|
EventsAfter(context.Context, int64, string, int64) ([]insight.Event, error)
|
|
LatestForPeriod(context.Context, int64, string, insight.Module, insight.PeriodScope) (insight.Snapshot, error)
|
|
RefreshForPeriod(context.Context, int64, string, insight.Module, insight.PeriodScope, string) (insight.Job, bool, error)
|
|
RefreshSectionForPeriod(context.Context, int64, string, insight.Module, insight.Section, insight.PeriodScope, string) (insight.Job, bool, error)
|
|
}
|
|
|
|
func (s *Server) refreshAIInsightSection(w http.ResponseWriter, r *http.Request) {
|
|
module, moduleOK := insight.ParseModule(r.PathValue("module"))
|
|
section, sectionOK := insight.ParseSection(module, r.PathValue("sectionId"))
|
|
if !moduleOK || s.insights == nil {
|
|
s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found")
|
|
return
|
|
}
|
|
if !sectionOK {
|
|
s.problem(w, r, 422, "VALIDATION_FAILED", "Unsupported AI insight section")
|
|
return
|
|
}
|
|
key := r.Header.Get("Idempotency-Key")
|
|
if !validIdempotencyKey(key) {
|
|
s.problem(w, r, 422, "VALIDATION_FAILED", "A valid Idempotency-Key is required")
|
|
return
|
|
}
|
|
var input struct {
|
|
TerminalID string `json:"terminalId"`
|
|
PeriodID string `json:"periodId"`
|
|
}
|
|
if err := decodeStrictJSON(w, r, &input); err != nil || len(input.TerminalID) > 128 {
|
|
s.problem(w, r, 422, "VALIDATION_FAILED", "Invalid insight refresh request")
|
|
return
|
|
}
|
|
period, periodOK := s.insightPeriod(r.Context(), module, input.PeriodID)
|
|
if !periodOK {
|
|
s.problem(w, r, 422, "VALIDATION_FAILED", "Unknown reporting period")
|
|
return
|
|
}
|
|
var job insight.Job
|
|
var cached bool
|
|
var err error
|
|
if period.PeriodID != "" {
|
|
job, cached, err = s.insights.RefreshSectionForPeriod(r.Context(), sessionFrom(r.Context()).MerchantID, input.TerminalID, module, section, period, key)
|
|
} else {
|
|
job, cached, err = s.insights.RefreshSection(r.Context(), sessionFrom(r.Context()).MerchantID, input.TerminalID, module, section, key)
|
|
}
|
|
if s.writeInsightRefreshResult(w, r, job, cached, err) {
|
|
return
|
|
}
|
|
}
|
|
|
|
func (s *Server) writeInsightRefreshResult(w http.ResponseWriter, r *http.Request, job insight.Job, cached bool, err error) bool {
|
|
switch {
|
|
case errors.Is(err, insight.ErrUnknownTerminal):
|
|
s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found")
|
|
return true
|
|
case errors.Is(err, insight.ErrIdempotencyConflict):
|
|
s.problem(w, r, 409, "IDEMPOTENCY_CONFLICT", "Idempotency key was already used for another request")
|
|
return true
|
|
case errors.Is(err, insight.ErrOnboardingRequired):
|
|
s.problem(w, r, 409, "ONBOARDING_REQUIRED", "Complete terminal onboarding before generating AI insights")
|
|
return true
|
|
case errors.Is(err, insight.ErrRefreshLimited), errors.Is(err, insight.ErrRefreshCooldown):
|
|
next := nextTehranDay(time.Now())
|
|
var cooldown *insight.RefreshCooldownError
|
|
if errors.As(err, &cooldown) {
|
|
next = cooldown.NextRefreshAt
|
|
}
|
|
w.Header().Set("Retry-After", strconv.FormatInt(max(1, int64(time.Until(next).Seconds())), 10))
|
|
w.Header().Set("Content-Type", "application/problem+json")
|
|
w.WriteHeader(429)
|
|
_ = json.NewEncoder(w).Encode(map[string]any{"type": "https://zarinpal.amiirkhl.ir/docs/errors/refresh-limited", "title": "Too Many Requests", "status": 429, "code": "REFRESH_LIMITED", "detail": "AI section refresh is not available yet", "instance": r.URL.Path, "requestId": w.Header().Get("X-Request-ID"), "nextRefreshAt": next})
|
|
return true
|
|
case err != nil:
|
|
s.logger.Error("creating AI insight refresh failed", "error", err)
|
|
s.problem(w, r, 503, "ANALYTICS_UNAVAILABLE", "AI insight unavailable")
|
|
return true
|
|
}
|
|
status := http.StatusAccepted
|
|
if cached {
|
|
status = http.StatusOK
|
|
}
|
|
w.Header().Set("Location", "/api/v1/analytics/ai-insight-refreshes/"+job.JobID)
|
|
s.writeJSON(w, status, job)
|
|
return true
|
|
}
|
|
|
|
func (s *Server) latestAIInsight(w http.ResponseWriter, r *http.Request) {
|
|
module, ok := insight.ParseModule(r.PathValue("module"))
|
|
if !ok || s.insights == nil {
|
|
s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found")
|
|
return
|
|
}
|
|
terminalID, periodID, ok := insightScopeQuery(r)
|
|
if !ok {
|
|
s.problem(w, r, 422, "VALIDATION_FAILED", "Only terminalId and periodId are accepted")
|
|
return
|
|
}
|
|
period, periodOK := s.insightPeriod(r.Context(), module, periodID)
|
|
if !periodOK {
|
|
s.problem(w, r, 422, "VALIDATION_FAILED", "Unknown reporting period")
|
|
return
|
|
}
|
|
var result insight.Snapshot
|
|
var err error
|
|
if period.PeriodID != "" {
|
|
result, err = s.insights.LatestForPeriod(r.Context(), sessionFrom(r.Context()).MerchantID, terminalID, module, period)
|
|
} else {
|
|
result, err = s.insights.Latest(r.Context(), sessionFrom(r.Context()).MerchantID, terminalID, module)
|
|
}
|
|
if errors.Is(err, insight.ErrNotFound) || errors.Is(err, insight.ErrUnknownTerminal) {
|
|
s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found")
|
|
return
|
|
}
|
|
if err != nil {
|
|
s.logger.Error("loading AI insight failed", "error", err, "module", module)
|
|
s.problem(w, r, 503, "ANALYTICS_UNAVAILABLE", "AI insight unavailable")
|
|
return
|
|
}
|
|
s.writeJSON(w, 200, result)
|
|
}
|
|
|
|
func (s *Server) refreshAIInsight(w http.ResponseWriter, r *http.Request) {
|
|
module, ok := insight.ParseModule(r.PathValue("module"))
|
|
if !ok || s.insights == nil {
|
|
s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found")
|
|
return
|
|
}
|
|
key := r.Header.Get("Idempotency-Key")
|
|
if !validIdempotencyKey(key) {
|
|
s.problem(w, r, 422, "VALIDATION_FAILED", "A valid Idempotency-Key is required")
|
|
return
|
|
}
|
|
var input struct {
|
|
TerminalID string `json:"terminalId"`
|
|
PeriodID string `json:"periodId"`
|
|
}
|
|
if err := decodeStrictJSON(w, r, &input); err != nil || len(input.TerminalID) > 128 {
|
|
s.problem(w, r, 422, "VALIDATION_FAILED", "Invalid insight refresh request")
|
|
return
|
|
}
|
|
period, periodOK := s.insightPeriod(r.Context(), module, input.PeriodID)
|
|
if !periodOK {
|
|
s.problem(w, r, 422, "VALIDATION_FAILED", "Unknown reporting period")
|
|
return
|
|
}
|
|
var job insight.Job
|
|
var cached bool
|
|
var err error
|
|
if period.PeriodID != "" {
|
|
job, cached, err = s.insights.RefreshForPeriod(r.Context(), sessionFrom(r.Context()).MerchantID, input.TerminalID, module, period, key)
|
|
} else {
|
|
job, cached, err = s.insights.Refresh(r.Context(), sessionFrom(r.Context()).MerchantID, input.TerminalID, module, key)
|
|
}
|
|
s.writeInsightRefreshResult(w, r, job, cached, err)
|
|
}
|
|
|
|
func (s *Server) getAIInsightJob(w http.ResponseWriter, r *http.Request) {
|
|
if s.insights == nil || !insightJobIDPattern.MatchString(r.PathValue("jobId")) {
|
|
s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found")
|
|
return
|
|
}
|
|
job, err := s.insights.GetJob(r.Context(), sessionFrom(r.Context()).MerchantID, r.PathValue("jobId"))
|
|
if errors.Is(err, insight.ErrNotFound) {
|
|
s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found")
|
|
return
|
|
}
|
|
if err != nil {
|
|
s.problem(w, r, 503, "ANALYTICS_UNAVAILABLE", "AI insight unavailable")
|
|
return
|
|
}
|
|
s.writeJSON(w, 200, job)
|
|
}
|
|
|
|
func (s *Server) aiInsightEvents(w http.ResponseWriter, r *http.Request) {
|
|
if s.insights == nil || !insightJobIDPattern.MatchString(r.PathValue("jobId")) {
|
|
s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found")
|
|
return
|
|
}
|
|
merchantID, jobID := sessionFrom(r.Context()).MerchantID, r.PathValue("jobId")
|
|
if _, err := s.insights.GetJob(r.Context(), merchantID, jobID); errors.Is(err, insight.ErrNotFound) {
|
|
s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found")
|
|
return
|
|
} else if err != nil {
|
|
s.problem(w, r, 503, "ANALYTICS_UNAVAILABLE", "AI insight unavailable")
|
|
return
|
|
}
|
|
flusher, ok := w.(http.Flusher)
|
|
if !ok {
|
|
s.problem(w, r, 500, "INTERNAL_ERROR", "Streaming unavailable")
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
|
|
w.Header().Set("Cache-Control", "no-cache, no-transform")
|
|
w.Header().Set("X-Accel-Buffering", "no")
|
|
last, _ := strconv.ParseInt(r.Header.Get("Last-Event-ID"), 10, 64)
|
|
ctx, cancel := context.WithTimeout(r.Context(), time.Hour)
|
|
defer cancel()
|
|
poll, heartbeat := time.NewTicker(2*time.Second), time.NewTicker(15*time.Second)
|
|
defer poll.Stop()
|
|
defer heartbeat.Stop()
|
|
for {
|
|
events, err := s.insights.EventsAfter(ctx, merchantID, jobID, last)
|
|
if err != nil {
|
|
return
|
|
}
|
|
for _, event := range events {
|
|
payload, _ := json.Marshal(event)
|
|
_, _ = fmt.Fprintf(w, "id: %d\nevent: %s\ndata: %s\n\n", event.ID, event.Type, payload)
|
|
last = event.ID
|
|
flusher.Flush()
|
|
}
|
|
job, err := s.insights.GetJob(ctx, merchantID, jobID)
|
|
if err != nil {
|
|
return
|
|
}
|
|
if job.State == "COMPLETED" || job.State == "FAILED" {
|
|
return
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-poll.C:
|
|
case <-heartbeat.C:
|
|
_, _ = fmt.Fprint(w, ": heartbeat\n\n")
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
}
|
|
|
|
func insightScopeQuery(r *http.Request) (string, string, bool) {
|
|
for key := range r.URL.Query() {
|
|
if key != "terminalId" && key != "periodId" {
|
|
return "", "", false
|
|
}
|
|
}
|
|
terminalID, periodID := r.URL.Query().Get("terminalId"), r.URL.Query().Get("periodId")
|
|
return terminalID, periodID, len(terminalID) <= 128 && len(periodID) <= 128
|
|
}
|
|
|
|
func (s *Server) insightPeriod(ctx context.Context, _ insight.Module, periodID string) (insight.PeriodScope, bool) {
|
|
if periodID == "" {
|
|
return insight.PeriodScope{}, true
|
|
}
|
|
if s.analytics == nil {
|
|
return insight.PeriodScope{}, false
|
|
}
|
|
periods, err := s.analytics.Periods(ctx)
|
|
if err != nil {
|
|
return insight.PeriodScope{}, false
|
|
}
|
|
for _, period := range periods {
|
|
if period.PeriodID == periodID {
|
|
return insight.PeriodScope{PeriodID: period.PeriodID, PeriodType: period.PeriodType, StartAt: period.StartAt, EndAt: period.EndAt, JalaliLabelFA: period.JalaliLabelFA}, true
|
|
}
|
|
}
|
|
return insight.PeriodScope{}, false
|
|
}
|
|
|
|
func nextTehranDay(now time.Time) time.Time {
|
|
location, err := time.LoadLocation("Asia/Tehran")
|
|
if err != nil {
|
|
location = time.FixedZone("Asia/Tehran", 3*60*60+30*60)
|
|
}
|
|
local := now.In(location)
|
|
return time.Date(local.Year(), local.Month(), local.Day()+1, 0, 0, 0, 0, location)
|
|
}
|