paliad/internal/services/dual_write.go

// Slice B.2 dual-write (t-paliad-305 / m/paliad#93) — keep paliad's
// new tables (procedural_events / sequencing_rules / legal_sources) in
// lock-step with the legacy paliad.deadline_rules table during the
// dual-write window. Mig 136 (Slice B.1) created the new tables and
// backfilled them once. This file keeps them in sync going forward.
//
// Contract:
//
//   - Every RuleEditorService method that mutates paliad.deadline_rules
//     calls syncDualWriteFromDeadlineRule(ctx, tx, id) inside the same
//     transaction, AFTER the deadline_rules write, BEFORE tx.Commit.
//   - The sync is idempotent (INSERT … ON CONFLICT … DO UPDATE) so the
//     same call works for Create (new row), UpdateDraft (existing row),
//     CloneAsDraft (new row referencing an old row), Publish (lifecycle
//     flip), Archive/Restore (lifecycle flip), and the published-peer
//     archive that Publish performs as a cascade.
//   - The sync re-derives the new-table state from paliad.deadline_rules
//     in pure SQL — no struct mapping in Go. The legacy table stays the
//     source of truth during B.2 (B.3 flips reads, B.4 drops it).
//   - Read paths still read deadline_rules in B.2. The new tables are a
//     parallel projection kept consistent for B.3's read cutover; they
//     are not yet authoritative.
//
// Why a per-row sync instead of a global trigger:
//
//   - The deadline_rules audit trigger (mig 079) reads paliad.audit_reason
//     to record the rationale on every change. Putting the new-table
//     write in the same TX preserves that auditability — set_config is
//     transactional and the new writes share the same reason.
//   - A Postgres-side AFTER UPDATE trigger on deadline_rules would also
//     work but it's harder to test in isolation and harder to revert
//     when B.4 drops the source table. A Go-side sync is reversible
//     with a code revert; an SQL trigger needs a follow-up migration.
//
// The drift-check job (CheckDualWriteDrift below) runs daily and
// alerts on mismatches. If the sync ever silently misses a row, the
// drift check surfaces it inside one day.
//
// See docs/design-procedural-events-model-2026-05-25.md §5.2 (dual-write
// phase) and docs/design-procedural-events-b0-findings-2026-05-26.md §7.
package services

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/google/uuid"
	"github.com/jmoiron/sqlx"
)

// syncDualWriteFromDeadlineRule re-projects the deadline_rules row with
// the given id into legal_sources + procedural_events + sequencing_rules.
// Runs three UPSERT statements in the open transaction.
//
// Synthetic-code rule (for rows where deadline_rules.submission_code is
// NULL) mirrors mig 136's backfill: 'null.' || first 8 hex chars of the
// uuid (dashes stripped). This must stay byte-identical to the mig 136
// expression or the lookup join inside the sequencing_rules UPSERT
// misses.
func syncDualWriteFromDeadlineRule(ctx context.Context, tx *sqlx.Tx, id uuid.UUID) error {
	// 1. legal_sources — UPSERT the citation (no-op if already present).
	//    jurisdiction is parsed from the first dot-separated segment;
	//    'other' on empty (paranoid fallback, no live rows hit it).
	if _, err := tx.ExecContext(ctx, `
		INSERT INTO paliad.legal_sources (citation, jurisdiction)
		SELECT dr.legal_source,
		       COALESCE(NULLIF(split_part(dr.legal_source, '.', 1), ''), 'other')
		  FROM paliad.deadline_rules dr
		 WHERE dr.id = $1 AND dr.legal_source IS NOT NULL
		ON CONFLICT (citation) DO NOTHING`, id); err != nil {
		return fmt.Errorf("dual-write legal_sources for rule %s: %w", id, err)
	}

	// 2. procedural_events — UPSERT keyed by code. The code is the
	//    submission_code if present, else the synthetic 'null.<8hex>'
	//    minted from the deadline_rules row's id (matches mig 136).
	//    legal_source_id is resolved by JOIN on legal_sources.citation
	//    (NULL when the rule has no legal_source).
	if _, err := tx.ExecContext(ctx, `
		INSERT INTO paliad.procedural_events
		    (code, name, name_en, description, event_kind,
		     primary_party_default, legal_source_id, concept_id,
		     lifecycle_state, published_at, is_active)
		SELECT
		    COALESCE(dr.submission_code,
		             'null.' || substring(replace(dr.id::text, '-', ''), 1, 8)),
		    dr.name, dr.name_en, dr.description, dr.event_type,
		    dr.primary_party, ls.id, dr.concept_id,
		    dr.lifecycle_state, dr.published_at, dr.is_active
		  FROM paliad.deadline_rules dr
		  LEFT JOIN paliad.legal_sources ls ON ls.citation = dr.legal_source
		 WHERE dr.id = $1
		ON CONFLICT (code) DO UPDATE SET
		    name                  = EXCLUDED.name,
		    name_en               = EXCLUDED.name_en,
		    description           = EXCLUDED.description,
		    event_kind            = EXCLUDED.event_kind,
		    primary_party_default = EXCLUDED.primary_party_default,
		    legal_source_id       = EXCLUDED.legal_source_id,
		    concept_id            = EXCLUDED.concept_id,
		    lifecycle_state       = EXCLUDED.lifecycle_state,
		    published_at          = EXCLUDED.published_at,
		    is_active             = EXCLUDED.is_active,
		    updated_at            = now()`, id); err != nil {
		return fmt.Errorf("dual-write procedural_events for rule %s: %w", id, err)
	}

	// 3. sequencing_rules — UPSERT keyed by id (1:1 inheritance from
	//    deadline_rules.id). procedural_event_id resolved by JOIN on
	//    the (real or synthetic) code. All hat-3 mechanics columns copy
	//    1:1 from the deadline_rules row's post-write state.
	if _, err := tx.ExecContext(ctx, `
		INSERT INTO paliad.sequencing_rules
		    (id, procedural_event_id, proceeding_type_id, parent_id, trigger_event_id,
		     duration_value, duration_unit, timing,
		     alt_duration_value, alt_duration_unit, alt_rule_code, anchor_alt,
		     combine_op, condition_expr, primary_party, sequence_order,
		     is_spawn, spawn_label, spawn_proceeding_type_id,
		     is_bilateral, is_court_set, priority,
		     rule_code, rule_codes, deadline_notes, deadline_notes_en,
		     choices_offered, applies_to_target,
		     lifecycle_state, draft_of, published_at, is_active,
		     created_at, updated_at)
		SELECT
		    dr.id, pe.id,
		    dr.proceeding_type_id, dr.parent_id, dr.trigger_event_id,
		    dr.duration_value, dr.duration_unit, dr.timing,
		    dr.alt_duration_value, dr.alt_duration_unit, dr.alt_rule_code, dr.anchor_alt,
		    dr.combine_op, dr.condition_expr, dr.primary_party, dr.sequence_order,
		    dr.is_spawn, dr.spawn_label, dr.spawn_proceeding_type_id,
		    dr.is_bilateral, dr.is_court_set, dr.priority,
		    dr.rule_code, dr.rule_codes, dr.deadline_notes, dr.deadline_notes_en,
		    dr.choices_offered, dr.applies_to_target,
		    dr.lifecycle_state, dr.draft_of, dr.published_at, dr.is_active,
		    dr.created_at, dr.updated_at
		  FROM paliad.deadline_rules dr
		  JOIN paliad.procedural_events pe
		    ON pe.code = COALESCE(dr.submission_code,
		                          'null.' || substring(replace(dr.id::text, '-', ''), 1, 8))
		 WHERE dr.id = $1
		ON CONFLICT (id) DO UPDATE SET
		    procedural_event_id      = EXCLUDED.procedural_event_id,
		    proceeding_type_id       = EXCLUDED.proceeding_type_id,
		    parent_id                = EXCLUDED.parent_id,
		    trigger_event_id         = EXCLUDED.trigger_event_id,
		    duration_value           = EXCLUDED.duration_value,
		    duration_unit            = EXCLUDED.duration_unit,
		    timing                   = EXCLUDED.timing,
		    alt_duration_value       = EXCLUDED.alt_duration_value,
		    alt_duration_unit        = EXCLUDED.alt_duration_unit,
		    alt_rule_code            = EXCLUDED.alt_rule_code,
		    anchor_alt               = EXCLUDED.anchor_alt,
		    combine_op               = EXCLUDED.combine_op,
		    condition_expr           = EXCLUDED.condition_expr,
		    primary_party            = EXCLUDED.primary_party,
		    sequence_order           = EXCLUDED.sequence_order,
		    is_spawn                 = EXCLUDED.is_spawn,
		    spawn_label              = EXCLUDED.spawn_label,
		    spawn_proceeding_type_id = EXCLUDED.spawn_proceeding_type_id,
		    is_bilateral             = EXCLUDED.is_bilateral,
		    is_court_set             = EXCLUDED.is_court_set,
		    priority                 = EXCLUDED.priority,
		    rule_code                = EXCLUDED.rule_code,
		    rule_codes               = EXCLUDED.rule_codes,
		    deadline_notes           = EXCLUDED.deadline_notes,
		    deadline_notes_en        = EXCLUDED.deadline_notes_en,
		    choices_offered          = EXCLUDED.choices_offered,
		    applies_to_target        = EXCLUDED.applies_to_target,
		    lifecycle_state          = EXCLUDED.lifecycle_state,
		    draft_of                 = EXCLUDED.draft_of,
		    published_at             = EXCLUDED.published_at,
		    is_active                = EXCLUDED.is_active,
		    updated_at               = now()`, id); err != nil {
		return fmt.Errorf("dual-write sequencing_rules for rule %s: %w", id, err)
	}

	return nil
}

// syncDeadlineDualLinks mirrors a deadline's legacy rule_id back-link
// onto the new procedural_event_id + sequencing_rule_id columns added
// by mig 136. Call this within an open transaction AFTER any UPDATE
// that mutates paliad.deadlines.rule_id (mig 122 introduced rule_id
// as the deadline→rule FK; today's writers are DeadlineService.Update
// and RuleEditorService.ResolveOrphan).
//
// Idempotent: NULL rule_id collapses both new columns to NULL by virtue
// of the subquery returning NULL. Slice B.2 (t-paliad-305).
func syncDeadlineDualLinks(ctx context.Context, tx *sqlx.Tx, deadlineID uuid.UUID) error {
	if _, err := tx.ExecContext(ctx, `
		UPDATE paliad.deadlines d
		   SET sequencing_rule_id  = d.rule_id,
		       procedural_event_id = (
		           SELECT sr.procedural_event_id
		             FROM paliad.sequencing_rules sr
		            WHERE sr.id = d.rule_id
		       )
		 WHERE d.id = $1`, deadlineID); err != nil {
		return fmt.Errorf("sync deadline dual-links for %s: %w", deadlineID, err)
	}
	return nil
}

// DualWriteDriftReport summarises the comparison between the legacy
// paliad.deadline_rules table and the new procedural_events /
// sequencing_rules tables that B.2's dual-write is meant to keep in
// sync. A zero-drift report (every count delta zero, every join clean)
// is the steady state during the dual-write window; any non-zero field
// is the signal that a write path either bypassed
// syncDualWriteFromDeadlineRule or that an out-of-band mutation
// happened (e.g. raw SQL run by an operator).
type DualWriteDriftReport struct {
	// Counts on the legacy and the projected side.
	DeadlineRules    int `json:"deadline_rules"`
	SequencingRules  int `json:"sequencing_rules"`
	ProceduralEvents int `json:"procedural_events"`
	LegalSources     int `json:"legal_sources"`

	// Expected (from the legacy side) vs observed (on the new side).
	ExpectedPE           int `json:"expected_procedural_events"`
	ExpectedLegalSources int `json:"expected_legal_sources"`

	// MissingSR — deadline_rules rows with no sequencing_rules row by id.
	// OrphanedSR — sequencing_rules rows whose id doesn't exist in
	// deadline_rules anymore (would only happen with a deletion path
	// that bypasses dual-write).
	MissingSR  int `json:"missing_sequencing_rules"`
	OrphanedSR int `json:"orphaned_sequencing_rules"`

	// MismatchedLifecycle — rows where deadline_rules.lifecycle_state
	// disagrees with sequencing_rules.lifecycle_state. Should always be
	// zero during dual-write.
	MismatchedLifecycle int `json:"mismatched_lifecycle"`

	// MismatchedActive — same shape, for is_active.
	MismatchedActive int `json:"mismatched_active"`
}

// HasDrift returns true if any field signals divergence between the
// legacy and projected sides. Used by the drift-check ticker to decide
// whether to log at WARN (drift) or INFO (clean).
func (r DualWriteDriftReport) HasDrift() bool {
	if r.SequencingRules != r.DeadlineRules {
		return true
	}
	if r.ProceduralEvents != r.ExpectedPE {
		return true
	}
	if r.LegalSources != r.ExpectedLegalSources {
		return true
	}
	if r.MissingSR != 0 || r.OrphanedSR != 0 {
		return true
	}
	if r.MismatchedLifecycle != 0 || r.MismatchedActive != 0 {
		return true
	}
	return false
}

// CheckDualWriteDrift compares the legacy paliad.deadline_rules table
// against the parallel new tables maintained by Slice B.2's dual-write.
// Returns a DualWriteDriftReport — caller decides what to do with
// non-zero drift (log, page, fail healthcheck, etc.).
//
// Read-only. Safe to run against prod. Single query per metric so the
// pool isn't held for a long time. No locks; tolerates concurrent
// writes (counts may shift by one or two during the read, but a
// persistent drift > 0 is the alarm signal).
func CheckDualWriteDrift(ctx context.Context, conn *sqlx.DB) (*DualWriteDriftReport, error) {
	var r DualWriteDriftReport

	q := func(label, sql string, dst *int) error {
		if err := conn.GetContext(ctx, dst, sql); err != nil {
			return fmt.Errorf("drift-check %s: %w", label, err)
		}
		return nil
	}

	if err := q("dr_total", `SELECT COUNT(*) FROM paliad.deadline_rules`, &r.DeadlineRules); err != nil {
		return nil, err
	}
	if err := q("sr_total", `SELECT COUNT(*) FROM paliad.sequencing_rules`, &r.SequencingRules); err != nil {
		return nil, err
	}
	if err := q("pe_total", `SELECT COUNT(*) FROM paliad.procedural_events`, &r.ProceduralEvents); err != nil {
		return nil, err
	}
	if err := q("ls_total", `SELECT COUNT(*) FROM paliad.legal_sources`, &r.LegalSources); err != nil {
		return nil, err
	}

	if err := q("expected_pe", `
		SELECT
		  (SELECT COUNT(DISTINCT submission_code) FROM paliad.deadline_rules WHERE submission_code IS NOT NULL)
		  +
		  (SELECT COUNT(*) FROM paliad.deadline_rules WHERE submission_code IS NULL)
	`, &r.ExpectedPE); err != nil {
		return nil, err
	}
	if err := q("expected_ls",
		`SELECT COUNT(DISTINCT legal_source) FROM paliad.deadline_rules WHERE legal_source IS NOT NULL`,
		&r.ExpectedLegalSources); err != nil {
		return nil, err
	}

	if err := q("missing_sr", `
		SELECT COUNT(*) FROM paliad.deadline_rules dr
		  LEFT JOIN paliad.sequencing_rules sr ON sr.id = dr.id
		 WHERE sr.id IS NULL`, &r.MissingSR); err != nil {
		return nil, err
	}
	if err := q("orphaned_sr", `
		SELECT COUNT(*) FROM paliad.sequencing_rules sr
		  LEFT JOIN paliad.deadline_rules dr ON dr.id = sr.id
		 WHERE dr.id IS NULL`, &r.OrphanedSR); err != nil {
		return nil, err
	}

	if err := q("mismatched_lifecycle", `
		SELECT COUNT(*) FROM paliad.deadline_rules dr
		  JOIN paliad.sequencing_rules sr ON sr.id = dr.id
		 WHERE dr.lifecycle_state <> sr.lifecycle_state`, &r.MismatchedLifecycle); err != nil {
		return nil, err
	}
	if err := q("mismatched_active", `
		SELECT COUNT(*) FROM paliad.deadline_rules dr
		  JOIN paliad.sequencing_rules sr ON sr.id = dr.id
		 WHERE dr.is_active <> sr.is_active`, &r.MismatchedActive); err != nil {
		return nil, err
	}

	return &r, nil
}

// StartDualWriteDriftCheckLoop runs CheckDualWriteDrift on a fixed
// interval for the lifetime of ctx. A clean run logs at INFO level;
// drift logs at WARN level with the full report payload. The first
// check fires after `interval`, not immediately on Start — by the time
// the ticker first fires the process has finished booting and the
// initial backfill + dual-write writes have settled.
//
// Slice B.2 (t-paliad-305). interval should be short enough to surface
// drift before the next deploy (so a broken dual-write doesn't sit
// silent for a week) and long enough to avoid noise (the check holds
// no locks but it does run nine SELECT COUNTs).
//
// Recommended interval: 6h. Override via the caller (cmd/server picks
// the runtime value).
func StartDualWriteDriftCheckLoop(ctx context.Context, conn *sqlx.DB, interval time.Duration) {
	if interval <= 0 {
		interval = 6 * time.Hour
	}
	go func() {
		t := time.NewTicker(interval)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				report, err := CheckDualWriteDrift(ctx, conn)
				if err != nil {
					log.Printf("dual-write drift-check: error: %v", err)
					continue
				}
				if report.HasDrift() {
					log.Printf("dual-write drift-check: DRIFT DETECTED — "+
						"deadline_rules=%d sequencing_rules=%d "+
						"procedural_events=%d (expected %d) "+
						"legal_sources=%d (expected %d) "+
						"missing_sr=%d orphaned_sr=%d "+
						"mismatched_lifecycle=%d mismatched_active=%d",
						report.DeadlineRules, report.SequencingRules,
						report.ProceduralEvents, report.ExpectedPE,
						report.LegalSources, report.ExpectedLegalSources,
						report.MissingSR, report.OrphanedSR,
						report.MismatchedLifecycle, report.MismatchedActive)
				} else {
					log.Printf("dual-write drift-check: OK — "+
						"deadline_rules=%d sequencing_rules=%d "+
						"procedural_events=%d legal_sources=%d",
						report.DeadlineRules, report.SequencingRules,
						report.ProceduralEvents, report.LegalSources)
				}
			}
		}
	}()
}