package redis

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

// Idempotency implements replay/coordination for idempotent mutations
// (migration-plan.md §19). Redis is coordination + replay only; PostgreSQL is
// authoritative. Record TTL bounds how long a completed response can be
// replayed; business invariants are never stored here.
//
// Record states:
//   - pending: a request claimed the key and is executing (or crashed).
//   - done:    the request completed; replay returns the stored response.
//
// A pending record older than staleAfter is treated as abandoned so a crashed
// handler cannot block the key forever.

const idemKeyPrefix = "idem:v1"

// IdemResult is a completed idempotent response ready to replay.
type IdemResult struct {
	Status      int
	Body        []byte
	ContentType string
}

// idemRecord is the JSON value stored per key.
type idemRecord struct {
	Status      string `json:"status"` // pending | done
	Fingerprint string `json:"fingerprint"`
	StatusCode  int    `json:"status_code,omitempty"`
	Body        string `json:"body,omitempty"`
	ContentType string `json:"content_type,omitempty"`
	CreatedAt   int64  `json:"created_at"`
}

// IdemKey builds the coordination key, scoped to the authenticated user, the
// normalized endpoint (method + route pattern) and the client idempotency key.
func IdemKey(userID int64, route, clientKey string) string {
	return fmt.Sprintf("%s:%d:%s:%s", idemKeyPrefix, userID, route, clientKey)
}

// claimScript atomically claims a pending slot or reports the existing state.
// Returns: "acquired" | "pending" | "conflict" | "replay"
var claimScript = redis.NewScript(`
local cur = redis.call('GET', KEYS[1])
if cur then
  local t = cjson.decode(cur)
  if t.fingerprint ~= ARGV[2] then
    return 'conflict'
  end
  if t.status == 'done' then return 'replay' end
  local age = tonumber(ARGV[4]) - t.created_at
  if age <= tonumber(ARGV[3]) then return 'pending' end
  -- stale pending slot with the same fingerprint: reclaim it (overwrite)
  redis.call('SET', KEYS[1],
    cjson.encode({status='pending', fingerprint=ARGV[2], created_at=tonumber(ARGV[4])}),
    'EX', ARGV[1])
  return 'acquired'
end
local ok = redis.call('SET', KEYS[1],
  cjson.encode({status='pending', fingerprint=ARGV[2], created_at=tonumber(ARGV[4])}),
  'NX', 'EX', ARGV[1])
if ok then return 'acquired' end
return 'pending'
`)

// Claim is the exported atomic claim entry point.
func Claim(ctx context.Context, rdb *redis.Client, key, fingerprint string, ttl, staleAfter time.Duration) (acquired bool, replay *IdemResult, err error) {
	return claim(ctx, rdb, key, fingerprint, ttl, staleAfter)
}

// Complete is the exported completion entry point.
func Complete(ctx context.Context, rdb *redis.Client, key, fingerprint string, ttl time.Duration, status int, body []byte, contentType string) error {
	return complete(ctx, rdb, key, fingerprint, ttl, status, body, contentType)
}

// Replay is the exported replay lookup entry point.
func Replay(ctx context.Context, rdb *redis.Client, key, fingerprint string) (*IdemResult, error) {
	return replayLookup(ctx, rdb, key, fingerprint)
}

// claim returns whether this request owns the key and may execute.
// ok=true means acquired (execute). replay carries a prior completed result.
func claim(ctx context.Context, rdb *redis.Client, key, fingerprint string, ttl, staleAfter time.Duration) (ok bool, replay *IdemResult, err error) {
	res, err := claimScript.Run(ctx, rdb, []string{key}, int64(ttl.Seconds()), fingerprint,
		int64(staleAfter.Seconds()), time.Now().Unix()).Result()
	if err != nil {
		return false, nil, err
	}
	switch res.(string) {
	case "acquired":
		return true, nil, nil
	case "replay":
		rep, err := fetchReplay(ctx, rdb, key)
		return false, rep, err
	case "conflict":
		return false, nil, ErrIdempotencyConflict
	case "pending":
		return false, nil, nil
	}
	return false, nil, fmt.Errorf("idempotency: unexpected claim result %q", res)
}

// complete stores a finished response for replay (only status < 500 is stored,
// so failed/infrastructure errors stay retryable).
func complete(ctx context.Context, rdb *redis.Client, key, fingerprint string, ttl time.Duration, status int, body []byte, contentType string) error {
	rec := idemRecord{Status: "done", Fingerprint: fingerprint, StatusCode: status,
		Body: string(body), ContentType: contentType, CreatedAt: time.Now().Unix()}
	b, err := json.Marshal(rec)
	if err != nil {
		return err
	}
	return rdb.Set(ctx, key, b, ttl).Err()
}

// replayLookup fetches a completed record and validates the fingerprint.
func replayLookup(ctx context.Context, rdb *redis.Client, key, fingerprint string) (*IdemResult, error) {
	raw, err := rdb.Get(ctx, key).Bytes()
	if err == redis.Nil {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	var rec idemRecord
	if err := json.Unmarshal(raw, &rec); err != nil {
		return nil, err
	}
	if rec.Status != "done" {
		return nil, nil
	}
	if rec.Fingerprint != fingerprint {
		return nil, ErrIdempotencyConflict
	}
	return &IdemResult{Status: rec.StatusCode, Body: []byte(rec.Body), ContentType: rec.ContentType}, nil
}

func fetchReplay(ctx context.Context, rdb *redis.Client, key string) (*IdemResult, error) {
	raw, err := rdb.Get(ctx, key).Bytes()
	if err == redis.Nil {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	var rec idemRecord
	if err := json.Unmarshal(raw, &rec); err != nil {
		return nil, err
	}
	if rec.Status != "done" {
		return nil, nil
	}
	return &IdemResult{Status: rec.StatusCode, Body: []byte(rec.Body), ContentType: rec.ContentType}, nil
}

// IdempotencyConflict marks a same-key/different-payload collision (§19).
var ErrIdempotencyConflict = fmt.Errorf("idempotency key already used with a different request")