// Package delivery implements the patterns that make distributed data
// changes safe: the transactional OUTBOX (solving the dual-write problem)
// and a crash-recoverable SAGA (multi-step workflows with compensation,
// instead of a distributed 2PC). Stdlib-only and deterministic.
package delivery

import "sync"

// Event is a domain event to be published (to a log, pa-04).
type Event struct {
	ID      string
	Type    string
	Payload string
}

// OutboxRow is an event stored atomically alongside the business write.
type OutboxRow struct {
	Seq       int
	Event     Event
	Published bool
}

// DB models a single service's database. The key move: AtomicWrite stores
// the business state change AND the outbox event in ONE transaction, so
// they can never disagree. (Here "transaction" = one mutex-guarded
// append; in production it's a real DB transaction.)
type DB struct {
	mu     sync.Mutex
	state  map[string]string
	outbox []*OutboxRow
	seq    int
}

func NewDB() *DB { return &DB{state: map[string]string{}} }

// AtomicWrite commits a state change and an outbox event together. This
// is the heart of the outbox pattern: no dual write, no lost events.
func (db *DB) AtomicWrite(key, val string, ev Event) {
	db.mu.Lock()
	defer db.mu.Unlock()
	db.state[key] = val
	db.outbox = append(db.outbox, &OutboxRow{Seq: db.seq, Event: ev})
	db.seq++
}

// DualWriteBroken demonstrates the ANTI-pattern: commit the state, then
// (separately) publish. If publish fails / the process crashes in
// between, the state changed but the event is lost — permanent
// inconsistency. crash=true skips the publish to simulate that.
func (db *DB) DualWriteBroken(key, val string, ev Event, publish func(Event) error, crash bool) error {
	db.mu.Lock()
	db.state[key] = val // committed
	db.mu.Unlock()
	if crash {
		return nil // crash before publish -> event LOST, state already changed
	}
	return publish(ev)
}

// Unpublished returns outbox rows not yet published (the relay's work).
func (db *DB) Unpublished() []*OutboxRow {
	db.mu.Lock()
	defer db.mu.Unlock()
	var out []*OutboxRow
	for _, r := range db.outbox {
		if !r.Published {
			out = append(out, r)
		}
	}
	return out
}

func (db *DB) markPublished(seq int) {
	db.mu.Lock()
	defer db.mu.Unlock()
	for _, r := range db.outbox {
		if r.Seq == seq {
			r.Published = true
		}
	}
}

func (db *DB) StateOf(key string) string {
	db.mu.Lock()
	defer db.mu.Unlock()
	return db.state[key]
}

// OutboxCount / StateCount let tests assert state and events never
// diverge (the outbox invariant).
func (db *DB) OutboxCount() int { db.mu.Lock(); defer db.mu.Unlock(); return len(db.outbox) }
func (db *DB) StateCount() int  { db.mu.Lock(); defer db.mu.Unlock(); return len(db.state) }

// Relay polls the outbox and publishes unpublished events to a sink, then
// marks them published. Because the publish and the mark are not atomic, a
// crash between them causes RE-publication on the next poll — i.e.
// at-least-once delivery, which is why the sink must be idempotent.
type Relay struct {
	db   *DB
	sink func(Event) error
}

func NewRelay(db *DB, sink func(Event) error) *Relay { return &Relay{db: db, sink: sink} }

// PollOnce publishes every unpublished row and marks it published.
func (r *Relay) PollOnce() (published int) {
	for _, row := range r.db.Unpublished() {
		if err := r.sink(row.Event); err != nil {
			continue // leave unpublished; retry next poll
		}
		r.db.markPublished(row.Seq)
		published++
	}
	return published
}

// PollOnceCrashBeforeMark publishes but does NOT mark — simulating a
// crash after the broker accepted the message but before the DB recorded
// it as sent. The next PollOnce will re-publish (the at-least-once case).
func (r *Relay) PollOnceCrashBeforeMark() (published int) {
	for _, row := range r.db.Unpublished() {
		if err := r.sink(row.Event); err == nil {
			published++
		}
	}
	return published
}

// IdempotentSink is a consumer that dedups by event id, so at-least-once
// re-delivery from the relay becomes effectively-once.
type IdempotentSink struct {
	mu         sync.Mutex
	seen       map[string]bool
	Delivered  int
	Duplicates int
}

func NewIdempotentSink() *IdempotentSink {
	return &IdempotentSink{seen: map[string]bool{}}
}

func (s *IdempotentSink) Deliver(ev Event) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.seen[ev.ID] {
		s.Duplicates++
		return nil
	}
	s.seen[ev.ID] = true
	s.Delivered++
	return nil
}