// Package outbox implements a durable transactional-outbox relay for AMQP
// publishing. Today the order service's amqp_emit hook publishes best-effort: a
// broker outage or a crash after a state change but before the publish loses
// the event. The outbox closes that gap.
//
// Producers call Publish, which appends the message to an append-only JSONL file
// and fsyncs before returning — so the intent to emit is durable the moment
// Publish returns, independent of broker availability. A single relay goroutine
// (Run) drains undelivered entries to the broker and records an ack line on
// success, retrying on failure. Delivery is therefore at-least-once: a crash
// between the broker publish and the ack line replays that message on restart,
// which downstream consumers must tolerate (the commerce events are keyed and
// already designed to be idempotent).
//
// This targets the single-node order service. The file is compacted on Open
// (acked entries are dropped), which bounds growth across restarts.
package outbox

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"log/slog"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"time"
)

// Publisher delivers a message to the broker. The relay retries on error, so an
// implementation should surface a transient failure (broker down) as an error
// rather than dropping the message.
type Publisher interface {
	Publish(exchange, routingKey string, body []byte) error
}

// line is one record in the log: either a pending message (Seq>0 with payload)
// or an ack marker (Ack>0). Body round-trips as base64 via encoding/json.
type line struct {
	Seq        uint64 `json:"seq,omitempty"`
	Exchange   string `json:"exchange,omitempty"`
	RoutingKey string `json:"rk,omitempty"`
	Body       []byte `json:"body,omitempty"`
	Ack        uint64 `json:"ack,omitempty"`
}

type entry struct {
	seq        uint64
	exchange   string
	routingKey string
	body       []byte
}

// Outbox is a durable, file-backed message buffer with an at-least-once relay.
type Outbox struct {
	mu      sync.Mutex
	f       *os.File
	path    string
	seq     uint64
	pending map[uint64]entry
	wake    chan struct{}
}

// Open loads the outbox at path (creating it if absent), drops already-acked
// entries, and rewrites the file with only the still-pending ones (compaction).
// Returns an Outbox whose pending set is ready for the relay to drain.
func Open(path string) (*Outbox, error) {
	pending := map[uint64]entry{}
	acked := map[uint64]struct{}{}
	var maxSeq uint64

	if f, err := os.Open(path); err == nil {
		sc := bufio.NewScanner(f)
		sc.Buffer(make([]byte, 0, 64*1024), 8*1024*1024)
		for sc.Scan() {
			b := sc.Bytes()
			if len(b) == 0 {
				continue
			}
			var l line
			if err := json.Unmarshal(b, &l); err != nil {
				continue // skip a torn/corrupt line
			}
			switch {
			case l.Ack > 0:
				acked[l.Ack] = struct{}{}
			case l.Seq > 0:
				pending[l.Seq] = entry{seq: l.Seq, exchange: l.Exchange, routingKey: l.RoutingKey, body: l.Body}
				if l.Seq > maxSeq {
					maxSeq = l.Seq
				}
			}
		}
		_ = f.Close()
	} else if !errors.Is(err, os.ErrNotExist) {
		return nil, err
	}
	for seq := range acked {
		delete(pending, seq)
	}

	ob := &Outbox{path: path, seq: maxSeq, pending: pending, wake: make(chan struct{}, 1)}
	if err := ob.compact(); err != nil {
		return nil, err
	}
	return ob, nil
}

// compact rewrites the file with only the current pending entries, then reopens
// it for appends. Called on Open so the log can't grow without bound across
// restarts. Must be called with no concurrent writers (Open, before Run).
func (o *Outbox) compact() error {
	tmp := o.path + ".tmp"
	f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644)
	if err != nil {
		return err
	}
	seqs := make([]uint64, 0, len(o.pending))
	for seq := range o.pending {
		seqs = append(seqs, seq)
	}
	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
	w := bufio.NewWriter(f)
	for _, seq := range seqs {
		e := o.pending[seq]
		b, _ := json.Marshal(line{Seq: e.seq, Exchange: e.exchange, RoutingKey: e.routingKey, Body: e.body})
		if _, err := w.Write(append(b, '\n')); err != nil {
			_ = f.Close()
			return err
		}
	}
	if err := w.Flush(); err != nil {
		_ = f.Close()
		return err
	}
	if err := f.Sync(); err != nil {
		_ = f.Close()
		return err
	}
	if err := f.Close(); err != nil {
		return err
	}
	if err := os.Rename(tmp, o.path); err != nil {
		return err
	}
	// fsync the directory so the rename (the file swap) is itself durable — a
	// crash right after must not leave the pending log lost or a stale .tmp.
	if d, derr := os.Open(filepath.Dir(o.path)); derr == nil {
		_ = d.Sync()
		_ = d.Close()
	}
	o.f, err = os.OpenFile(o.path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
	return err
}

// Publish durably records a message and returns once it is fsync'd. It satisfies
// the order.Publisher seam, so the amqp_emit hook records to the outbox instead
// of publishing to the broker directly. The relay delivers it asynchronously.
func (o *Outbox) Publish(exchange, routingKey string, body []byte) error {
	o.mu.Lock()
	o.seq++
	seq := o.seq
	b, err := json.Marshal(line{Seq: seq, Exchange: exchange, RoutingKey: routingKey, Body: body})
	if err != nil {
		o.mu.Unlock()
		return err
	}
	if _, err := o.f.Write(append(b, '\n')); err != nil {
		o.mu.Unlock()
		return err
	}
	if err := o.f.Sync(); err != nil {
		o.mu.Unlock()
		return err
	}
	o.pending[seq] = entry{seq: seq, exchange: exchange, routingKey: routingKey, body: body}
	o.mu.Unlock()

	select {
	case o.wake <- struct{}{}: // nudge the relay
	default:
	}
	return nil
}

// PendingCount reports how many messages are awaiting delivery (for tests/metrics).
func (o *Outbox) PendingCount() int {
	o.mu.Lock()
	defer o.mu.Unlock()
	return len(o.pending)
}

// Run drains the outbox to pub, retrying failed rounds every retry interval and
// waking immediately when a new message arrives. It blocks until ctx is done.
func (o *Outbox) Run(ctx context.Context, pub Publisher, retry time.Duration, logger *slog.Logger) {
	t := time.NewTicker(retry)
	defer t.Stop()
	for {
		o.drain(pub, logger)
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		case <-o.wake:
		}
	}
}

// drain attempts to deliver every pending message in seq order. It stops the
// round at the first delivery error (likely the broker is down) so messages keep
// their order and the next tick retries.
func (o *Outbox) drain(pub Publisher, logger *slog.Logger) {
	o.mu.Lock()
	seqs := make([]uint64, 0, len(o.pending))
	for seq := range o.pending {
		seqs = append(seqs, seq)
	}
	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
	batch := make([]entry, 0, len(seqs))
	for _, seq := range seqs {
		batch = append(batch, o.pending[seq])
	}
	o.mu.Unlock()

	for _, e := range batch {
		if err := pub.Publish(e.exchange, e.routingKey, e.body); err != nil {
			if logger != nil {
				logger.Warn("outbox: delivery failed, will retry", "seq", e.seq, "err", err)
			}
			return // broker likely down; retry the whole round next tick
		}
		if err := o.ack(e.seq); err != nil && logger != nil {
			// The message was delivered; failing to persist the ack only risks a
			// redelivery on restart (at-least-once), so log and continue.
			logger.Warn("outbox: ack write failed", "seq", e.seq, "err", err)
		}
	}
}

// ack records that seq was delivered and drops it from the pending set.
func (o *Outbox) ack(seq uint64) error {
	o.mu.Lock()
	defer o.mu.Unlock()
	b, _ := json.Marshal(line{Ack: seq})
	if _, err := o.f.Write(append(b, '\n')); err != nil {
		return err
	}
	if err := o.f.Sync(); err != nil {
		return err
	}
	delete(o.pending, seq)
	return nil
}

// Close closes the underlying file.
func (o *Outbox) Close() error {
	o.mu.Lock()
	defer o.mu.Unlock()
	return o.f.Close()
}