// Package redis implements one managed Redis Streams consumer group per queue.
// Use Redis 6.2+, persistence, noeviction and a private namespace. The application
// owns the client. Delivery durability depends on the Redis deployment.
package redis

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"git.yuntuo.hk/k2safe/go-omni/queue"
	goredis "github.com/redis/go-redis/v9"
)

type Config struct {
	Namespace       string        `json:"namespace"`
	MaxRecords      int           `json:"maxRecords"` // Includes pending, dead and retained completed identities.
	MaxPayloadBytes int           `json:"maxPayloadBytes"`
	DedupTTL        time.Duration `json:"dedupTtl"` // Starts at ACK; pending/dead records never expire automatically.
}
type Store struct {
	client    goredis.UniversalClient
	cfg       Config
	keys      []string
	signature string
}

func New(client goredis.UniversalClient, cfg Config) (*Store, error) {
	if client == nil || cfg.Namespace == "" || len(cfg.Namespace) > 256 || cfg.MaxRecords < 0 || cfg.MaxPayloadBytes < 0 || cfg.DedupTTL < 0 {
		return nil, errors.New("queue/redis: invalid configuration")
	}
	if cfg.MaxRecords == 0 {
		cfg.MaxRecords = 1000
	}
	if cfg.MaxPayloadBytes == 0 {
		cfg.MaxPayloadBytes = 64 << 10
	}
	if cfg.DedupTTL == 0 {
		cfg.DedupTTL = 24 * time.Hour
	}
	if cfg.MaxRecords > 100000 || cfg.MaxPayloadBytes > 1<<20 || cfg.DedupTTL < time.Second || cfg.DedupTTL > 30*24*time.Hour {
		return nil, errors.New("queue/redis: configuration exceeds supported bounds")
	}
	sum := sha256.Sum256([]byte(cfg.Namespace))
	base := "goOmni:{" + hex.EncodeToString(sum[:]) + "}:queue"
	return &Store{client: client, cfg: cfg, keys: []string{base + ":stream", base + ":records", base + ":expiry", base + ":dead", base + ":config", base + ":consumerConfig"}, signature: fmt.Sprintf("%d:%d:%d", cfg.MaxRecords, cfg.MaxPayloadBytes, cfg.DedupTTL.Milliseconds())}, nil
}

var publish = goredis.NewScript(`
local config=redis.call('GET',KEYS[5]);if config and config~=ARGV[5] then return redis.error_reply('configuration mismatch') end
if not config then redis.call('SET',KEYS[5],ARGV[5]) end
local clock=redis.call('TIME');local now=clock[1]*1000+math.floor(clock[2]/1000)
local expired=redis.call('ZRANGEBYSCORE',KEYS[3],'-inf',now,'LIMIT',0,128)
for _,id in ipairs(expired) do redis.call('HDEL',KEYS[2],id);redis.call('ZREM',KEYS[3],id) end
local expiry=redis.call('ZSCORE',KEYS[3],ARGV[1]);if expiry and tonumber(expiry)<=now then redis.call('HDEL',KEYS[2],ARGV[1]);redis.call('ZREM',KEYS[3],ARGV[1]) end
local old=redis.call('HGET',KEYS[2],ARGV[1]);if old then local r=cjson.decode(old);if r.fingerprint~=ARGV[2] then return {-1,''} end;return {0,r.streamId} end
if redis.call('HLEN',KEYS[2])>=tonumber(ARGV[4]) then return {-2,''} end
local id=redis.call('XADD',KEYS[1],'*','message',ARGV[3])
redis.call('HSET',KEYS[2],ARGV[1],cjson.encode({fingerprint=ARGV[2],streamId=id,state='pending'}));return {1,id}
`)

func (s *Store) Publish(ctx context.Context, m queue.Message) (queue.Receipt, error) {
	if err := queue.Validate(ctx, m, s.cfg.MaxPayloadBytes); err != nil {
		return queue.Receipt{}, err
	}
	raw, _ := json.Marshal(m)
	if len(raw) > s.cfg.MaxPayloadBytes+1024 {
		return queue.Receipt{}, errors.New("queue/redis: encoded message exceeds limit")
	}
	sum := sha256.Sum256(raw)
	r, err := publish.Run(ctx, s.client, s.keys, m.ID, hex.EncodeToString(sum[:]), string(raw), s.cfg.MaxRecords, s.signature).Slice()
	if err != nil {
		return queue.Receipt{}, err
	}
	if len(r) != 2 {
		return queue.Receipt{}, errors.New("queue/redis: invalid publish result")
	}
	switch r[0] {
	case int64(-1):
		return queue.Receipt{}, queue.ErrConflict
	case int64(-2):
		return queue.Receipt{}, queue.ErrFull
	}
	id, ok := r[1].(string)
	if !ok {
		return queue.Receipt{}, errors.New("queue/redis: invalid receipt")
	}
	return queue.Receipt{ID: m.ID, BackendID: id}, nil
}

var finish = goredis.NewScript(`
local p=redis.call('XPENDING',KEYS[1],'workers',ARGV[1],ARGV[1],1)
if #p~=1 or p[1][2]~=ARGV[2] then return 0 end
local raw=redis.call('HGET',KEYS[2],ARGV[3]);if not raw then return redis.error_reply('message record missing') end
local record=cjson.decode(raw);if record.streamId~=ARGV[1] or record.state~='pending' then return redis.error_reply('message identity mismatch') end
if ARGV[4]~='' then
 record.state='dead';redis.call('ZADD',KEYS[4],0,ARGV[1]);record.dead=cjson.decode(ARGV[4])
else
 record.state='completed';local clock=redis.call('TIME');local now=clock[1]*1000+math.floor(clock[2]/1000);redis.call('ZADD',KEYS[3],now+tonumber(ARGV[5]),ARGV[3])
end
redis.call('HSET',KEYS[2],ARGV[3],cjson.encode(record));redis.call('XACK',KEYS[1],'workers',ARGV[1])
if ARGV[4]=='' then redis.call('XDEL',KEYS[1],ARGV[1]) end
return 1
`)

func (s *Store) finish(ctx context.Context, id, owner string, m queue.Delivery, reason string) error {
	var dead []byte
	if reason != "" {
		dead, _ = json.Marshal(map[string]any{"attempt": m.Attempt, "reason": reason})
	}
	n, err := finish.Run(ctx, s.client, s.keys, id, owner, m.ID, string(dead), s.cfg.DedupTTL.Milliseconds()).Int()
	if err != nil {
		return err
	}
	if n != 1 {
		return queue.ErrLeaseLost
	}
	return nil
}

type DeadLetter struct {
	StreamID string        `json:"streamId"`
	Message  queue.Message `json:"message"`
	Attempt  int64         `json:"attempt"`
	Reason   string        `json:"reason"`
}

// DeadLetters returns a bounded page, ordered by opaque Redis stream ID. Pass
// the last StreamID as after. Payloads remain retained until explicit removal.
func (s *Store) DeadLetters(ctx context.Context, after string, limit int64) ([]DeadLetter, error) {
	if ctx == nil || limit < 1 || limit > 100 {
		return nil, errors.New("queue/redis: context and limit 1..100 required")
	}
	min := "-"
	if after != "" {
		min = "(" + after
	}
	ids, err := s.client.ZRangeByLex(ctx, s.keys[3], &goredis.ZRangeBy{Min: min, Max: "+", Count: limit}).Result()
	if err != nil {
		return nil, err
	}
	out := make([]DeadLetter, 0, len(ids))
	for _, id := range ids {
		messages, err := s.client.XRangeN(ctx, s.keys[0], id, id, 1).Result()
		if err != nil {
			return nil, err
		}
		if len(messages) == 0 {
			continue
		}
		var m queue.Message
		raw, _ := messages[0].Values["message"].(string)
		if err = json.Unmarshal([]byte(raw), &m); err != nil {
			return nil, err
		}
		record, err := s.client.HGet(ctx, s.keys[1], m.ID).Result()
		if err == goredis.Nil {
			continue
		}
		if err != nil {
			return nil, err
		}
		var saved struct {
			Dead struct {
				Attempt int64  `json:"attempt"`
				Reason  string `json:"reason"`
			} `json:"dead"`
		}
		if err = json.Unmarshal([]byte(record), &saved); err != nil {
			return nil, err
		}
		out = append(out, DeadLetter{StreamID: id, Message: m, Attempt: saved.Dead.Attempt, Reason: saved.Dead.Reason})
	}
	return out, nil
}

// DiscardDead deletes only a matching dead delivery after operator review.
// It retains the producer identity for DedupTTL, preventing immediate resubmit.
func (s *Store) DiscardDead(ctx context.Context, messageID, streamID string) error {
	return goredis.NewScript(`local raw=redis.call('HGET',KEYS[2],ARGV[1]);if not raw then return 0 end;local r=cjson.decode(raw);if r.state~='dead' or r.streamId~=ARGV[2] then return redis.error_reply('dead identity mismatch') end;r.state='discarded';r.dead=nil;redis.call('HSET',KEYS[2],ARGV[1],cjson.encode(r));redis.call('ZREM',KEYS[4],ARGV[2]);redis.call('XDEL',KEYS[1],ARGV[2]);local t=redis.call('TIME');redis.call('ZADD',KEYS[3],t[1]*1000+math.floor(t[2]/1000)+tonumber(ARGV[3]),ARGV[1]);return 1`).Run(ctx, s.client, s.keys, messageID, streamID, s.cfg.DedupTTL.Milliseconds()).Err()
}

var _ queue.Publisher = (*Store)(nil)