package main

// 🤖 Blob mirroring + audit (federated blobs Phase A, #992 —
// docs/federated-blob-storage.md): a second S3-compatible store holds a copy
// of every store-backed blob, and both stores are spot-checked against the
// sha256 the upload recorded. Write-once ciphertext makes all of this
// embarrassingly simple: the mirror sweep is idempotent and order-free (the
// blobs table IS the journal — no state machine, no lease), and a re-run
// after any crash converges.
//
// Enabled per instance by the LCHAT_BLOB_S3_MIRROR_* env quartet (+ key
// pair) beside the primary's; requires the primary S3 store (a mirror of the
// bare disk layout is the fleet backup's job, not this one's). Reads chain
// primary → mirror → disk, so a store outage degrades to slower reads, never
// errors. Deletes reap both stores. Changing the mirror TARGET later means
// resetting the watermark: UPDATE blobs SET mirrored_at=0 — documented beside
// the env reference; Phase B's store registry replaces this with per-store
// placement rows.

import (
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"io"
	"io/fs"
	"log"
	"time"
)

// mirroredBlobStore composes the primary (which itself falls back to disk for
// pre-flip blobs) with the mirror. The store interface stays the seam: the
// sweep below copies through create/commit, so nothing here knows S3.
type mirroredBlobStore struct {
	primary blobStore
	mirror  blobStore
}

func (m *mirroredBlobStore) create() (blobWriter, error) { return m.primary.create() }

func (m *mirroredBlobStore) open(id string) (io.ReadSeekCloser, int64, time.Time, error) {
	r, size, mod, err := m.primary.open(id)
	if err == nil {
		return r, size, mod, nil
	}
	if errors.Is(err, fs.ErrNotExist) {
		// Primary outage looks like errors, not not-founds; a genuine miss
		// (a lost primary object) served from the mirror is the redundancy
		// paying out — loudly, because it means the primary needs healing.
		if r2, s2, m2, err2 := m.mirror.open(id); err2 == nil {
			log.Printf("blobs: %s served from the MIRROR (primary miss — investigate)", id)
			return r2, s2, m2, nil
		}
	}
	return nil, 0, time.Time{}, err
}

func (m *mirroredBlobStore) remove(id string) error {
	err := m.primary.remove(id)
	if err2 := m.mirror.remove(id); err == nil {
		err = err2
	}
	return err
}

// ---- the mirror sweep -------------------------------------------------------

// unmirroredBlobs returns store-backed rows the mirror lacks, oldest first.
// Legacy in-DB rows (bytes in the data column) are excluded by the
// length(data)=0 predicate — they ride the DB's own replication.
func (db *DB) unmirroredBlobs(limit int) ([]struct{ ID, SHA string }, error) {
	rows, err := db.sql.Query(
		`SELECT id, sha256 FROM blobs
		 WHERE mirrored_at = 0 AND length(data) = 0
		 ORDER BY created_at LIMIT ?`, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []struct{ ID, SHA string }
	for rows.Next() {
		var r struct{ ID, SHA string }
		if err := rows.Scan(&r.ID, &r.SHA); err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

func (db *DB) markMirrored(id, sha256Hex string) error {
	_, err := db.sql.Exec(
		"UPDATE blobs SET mirrored_at = ?, sha256 = ? WHERE id = ?",
		time.Now().Unix(), sha256Hex, id)
	return err
}

// mirrorOnce copies one batch of unmirrored blobs primary→mirror, hashing in
// transit: rows without a sha256 (pre-hash uploads) get it backfilled — the
// sweep reads the full bytes anyway — and rows WITH one are verified, so a
// primary that corrupted a blob is caught before the corruption spreads to
// the copy. Returns how many blobs were mirrored this pass.
func (db *DB) mirrorOnce(batch int) (int, error) {
	m, ok := db.blobs.(*mirroredBlobStore)
	if !ok {
		return 0, nil
	}
	pending, err := db.unmirroredBlobs(batch)
	if err != nil {
		return 0, err
	}
	done := 0
	for _, row := range pending {
		r, _, _, err := m.primary.open(row.ID)
		if err != nil {
			if errors.Is(err, fs.ErrNotExist) {
				// Deleted since the query (GC won the race) — the row is gone
				// or going; either way the next pass won't see it. A row that
				// PERSISTS with no bytes is a durability incident the audit
				// pass reports.
				continue
			}
			return done, err
		}
		w, err := m.mirror.create()
		if err != nil {
			r.Close()
			return done, err
		}
		hasher := sha256.New()
		_, copyErr := io.Copy(io.MultiWriter(w, hasher), r)
		r.Close()
		if copyErr != nil {
			w.abort()
			return done, copyErr
		}
		sum := hex.EncodeToString(hasher.Sum(nil))
		if row.SHA != "" && sum != row.SHA {
			// The PRIMARY's bytes don't match what was uploaded. Never mirror
			// corruption — leave the row unmirrored and shout; the mirror may
			// be about to become the only good copy there is.
			w.abort()
			log.Printf("blobs: INTEGRITY: primary bytes for %s hash %.12s, recorded %.12s — not mirrored", row.ID, sum, row.SHA)
			continue
		}
		if err := w.commit(row.ID); err != nil {
			return done, err
		}
		if err := db.markMirrored(row.ID, sum); err != nil {
			return done, err
		}
		done++
	}
	return done, nil
}

// ---- the audit sweep --------------------------------------------------------

// blobAuditResult is one pass's verdict, per store — returned (not just
// logged) so tests can pin the behavior and Phase B can publish it.
type blobAuditResult struct {
	Checked, BadPrimary, BadMirror, MissingPrimary, MissingMirror int
}

// sampleHashedBlobs picks up to k random mirrored, hashed rows to audit.
func (db *DB) sampleHashedBlobs(k int) (map[string]string, error) {
	rows, err := db.sql.Query(
		`SELECT id, sha256 FROM blobs
		 WHERE sha256 != '' AND mirrored_at > 0 AND length(data) = 0
		 ORDER BY RANDOM() LIMIT ?`, k)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := map[string]string{}
	for rows.Next() {
		var id, sha string
		if err := rows.Scan(&id, &sha); err != nil {
			return nil, err
		}
		out[id] = sha
	}
	return out, rows.Err()
}

func hashFromStore(store blobStore, id string) (string, error) {
	r, _, _, err := store.open(id)
	if err != nil {
		return "", err
	}
	defer r.Close()
	hasher := sha256.New()
	if _, err := io.Copy(hasher, r); err != nil {
		return "", err
	}
	return hex.EncodeToString(hasher.Sum(nil)), nil
}

// auditOnce full-reads k sampled blobs from EACH store and verifies them
// against the recorded hash — probabilistic coverage over time at a few
// requests per pass (docs/federated-blob-storage.md § audit sweep). Mismatches
// and misses are loud: an integrity failure on either store is the signal the
// whole trust-but-verify model hangs on.
func (db *DB) auditBlobsOnce(k int) (blobAuditResult, error) {
	var res blobAuditResult
	m, ok := db.blobs.(*mirroredBlobStore)
	if !ok {
		return res, nil
	}
	sample, err := db.sampleHashedBlobs(k)
	if err != nil {
		return res, err
	}
	check := func(store blobStore, id, want string, bad, missing *int, name string) {
		got, err := hashFromStore(store, id)
		switch {
		case errors.Is(err, fs.ErrNotExist):
			*missing++
			log.Printf("blobs: AUDIT: %s MISSING from %s", id, name)
		case err != nil:
			// Transient store trouble isn't an integrity verdict — skip, the
			// next pass re-samples.
		case got != want:
			*bad++
			log.Printf("blobs: AUDIT: %s CORRUPT on %s (hash %.12s, recorded %.12s)", id, name, got, want)
		}
	}
	for id, want := range sample {
		res.Checked++
		check(m.primary, id, want, &res.BadPrimary, &res.MissingPrimary, "primary")
		check(m.mirror, id, want, &res.BadMirror, &res.MissingMirror, "mirror")
	}
	return res, nil
}

// ---- the loops ---------------------------------------------------------------

// blobMirrorKick nudges the sweep after an upload so a fresh blob mirrors
// within seconds while the space is awake (a hibernating space's unmirrored
// tail waits for its next wake — the honest Phase A window).
func (s *Server) kickBlobMirror() {
	if s.blobMirrorKick == nil {
		return
	}
	select {
	case s.blobMirrorKick <- struct{}{}:
	default: // a kick is already pending — the sweep drains everything anyway
	}
}

// startBlobMirror launches the mirror + audit loops when the store is
// mirrored. Cadences: the mirror ticks every 5m (plus upload kicks) and
// drains in batches; the audit samples a handful hourly — both idle at zero
// S3 requests when there's nothing to do (#836's rule).
func (s *Server) startBlobMirror() {
	if _, ok := s.db.blobs.(*mirroredBlobStore); !ok {
		return
	}
	s.blobMirrorKick = make(chan struct{}, 1)
	go func() {
		tick := time.NewTicker(5 * time.Minute)
		defer tick.Stop()
		for {
			select {
			case <-tick.C:
			case <-s.blobMirrorKick:
			}
			for {
				n, err := s.db.mirrorOnce(32)
				if err != nil {
					log.Printf("blobs: mirror sweep: %v (retrying next tick)", err)
					break
				}
				if n < 32 {
					break // drained
				}
			}
		}
	}()
	go func() {
		tick := time.NewTicker(1 * time.Hour)
		defer tick.Stop()
		for range tick.C {
			if res, err := s.db.auditBlobsOnce(3); err != nil {
				log.Printf("blobs: audit sweep: %v", err)
			} else if res.BadPrimary+res.BadMirror+res.MissingPrimary+res.MissingMirror > 0 {
				log.Printf("blobs: AUDIT FAILURES this pass: %+v", res)
			}
		}
	}()
	log.Printf("blobs: mirror + audit sweeps on")
}