zarinpal-analytics-backend/internal/api/insights.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)
	GetJob(context.Context, int64, string) (insight.Job, error)
	EventsAfter(context.Context, int64, string, int64) ([]insight.Event, error)
}

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, ok := insightTerminalQuery(r)
	if !ok {
		s.problem(w, r, 422, "VALIDATION_FAILED", "Only terminalId is accepted")
		return
	}
	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"`
	}
	if err := decodeStrictJSON(w, r, &input); err != nil || len(input.TerminalID) > 128 {
		s.problem(w, r, 422, "VALIDATION_FAILED", "Invalid insight refresh request")
		return
	}
	job, cached, err := s.insights.Refresh(r.Context(), sessionFrom(r.Context()).MerchantID, input.TerminalID, module, key)
	switch {
	case errors.Is(err, insight.ErrUnknownTerminal):
		s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found")
		return
	case errors.Is(err, insight.ErrIdempotencyConflict):
		s.problem(w, r, 409, "IDEMPOTENCY_CONFLICT", "Idempotency key was already used for another request")
		return
	case errors.Is(err, insight.ErrRefreshLimited):
		next := nextTehranDay(time.Now())
		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": "Daily refresh is not available yet", "instance": r.URL.Path, "requestId": w.Header().Get("X-Request-ID"), "nextRefreshAt": next})
		return
	case err != nil:
		s.logger.Error("creating AI insight refresh failed", "error", err)
		s.problem(w, r, 503, "ANALYTICS_UNAVAILABLE", "AI insight unavailable")
		return
	}
	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)
}

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 insightTerminalQuery(r *http.Request) (string, bool) {
	for key := range r.URL.Query() {
		if key != "terminalId" {
			return "", false
		}
	}
	value := r.URL.Query().Get("terminalId")
	return value, len(value) <= 128
}

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)
}