paliad/internal/services/backup_service_test.go

package services

// Pure-function tests for the Backup Mode runtime (t-paliad-246 / m/paliad#77).
//
// Live DB behaviour (the actual org dump end-to-end) needs a Postgres;
// it would live in backup_service_live_test.go under TEST_DATABASE_URL.
// This file covers the bits that don't need a database:
//
//   - orgSheetSpecs registry shape: no duplicates, no excluded
//     paliadin sheets, predictable prefix split between entity and ref.
//   - composeOrgSheetSQL drift-resistance: missing ORDER BY cols drop,
//     SQL override path bypasses the builder, all-missing → no clause.
//   - LocalDiskStore Put / Get / Delete round-trip, key validation,
//     URI traversal rejection.

import (
	"bytes"
	"context"
	"io"
	"os"
	"path/filepath"
	"strings"
	"testing"
)

// ---------------------------------------------------------------------------
// orgSheetSpecs registry
// ---------------------------------------------------------------------------

func TestOrgSheetSpecs_NoDuplicates(t *testing.T) {
	seen := map[string]bool{}
	for _, sp := range orgSheetSpecs() {
		if seen[sp.SheetName] {
			t.Fatalf("duplicate sheet name in orgSheetSpecs: %q", sp.SheetName)
		}
		seen[sp.SheetName] = true
	}
}

func TestOrgSheetSpecs_ExcludesPaliadinTables(t *testing.T) {
	// m's t-paliad-214 Q5 decision + this design's §11 Q3 default:
	// paliadin_turns and paliadin_aichat_conversation must be ABSENT
	// from the registry (structural exclusion, not just column-drop).
	for _, sp := range orgSheetSpecs() {
		name := sp.SheetName
		if strings.Contains(name, "paliadin") {
			t.Fatalf("orgSheetSpecs leaked paliadin sheet: %q (m's Q3 mandates structural exclusion)", name)
		}
		if strings.Contains(sp.Table, "paliadin") {
			t.Fatalf("orgSheetSpecs[%q].Table references a paliadin table: %s", name, sp.Table)
		}
		// Belt-and-braces: SQL override bodies (the few sheets that
		// bypass the Table+OrderBy builder) also can't pull paliadin
		// tables in through UNION/subquery.
		if strings.Contains(sp.SQL, "paliadin_turns") || strings.Contains(sp.SQL, "paliadin_aichat_conversation") {
			t.Fatalf("orgSheetSpecs[%q] SQL references a paliadin table: %s", name, sp.SQL)
		}
	}
}

func TestOrgSheetSpecs_RefSheetsPrefixed(t *testing.T) {
	// Every sheet whose data is read-only reference material is
	// expected to use the `ref__` prefix. The writer's downstream
	// consumers rely on this convention to group reference data
	// visually in the workbook.
	for _, sp := range orgSheetSpecs() {
		if !strings.HasPrefix(sp.SheetName, "ref__") {
			continue
		}
		// Reference sheets shouldn't carry per-row WHERE clauses (they
		// dump the whole reference table for portability). Only
		// applies to the SQL-override path; the Table+OrderBy builder
		// never emits a WHERE.
		if sp.SQL != "" && strings.Contains(strings.ToUpper(sp.SQL), "WHERE") {
			t.Fatalf("orgSheetSpecs[%q] is ref__ but has a WHERE clause; reference sheets dump the whole table", sp.SheetName)
		}
	}
}

func TestOrgSheetSpecs_OrderByForDeterminism(t *testing.T) {
	// Every sheet must declare a stable sort: either OrderBy on the
	// Table+OrderBy path, or ORDER BY in the SQL override. Keeps the
	// byte-deterministic contract from t-paliad-214 §3 across runs.
	//
	// (Drift removes ORDER BY columns at runtime, but only ones that
	// no longer exist in the schema — the spec-level declaration is
	// still required so we know what *should* be ordered.)
	for _, sp := range orgSheetSpecs() {
		if sp.SQL != "" {
			if !strings.Contains(strings.ToUpper(sp.SQL), "ORDER BY") {
				t.Fatalf("orgSheetSpecs[%q] SQL override missing ORDER BY (determinism contract): %s", sp.SheetName, sp.SQL)
			}
			continue
		}
		if len(sp.OrderBy) == 0 {
			t.Fatalf("orgSheetSpecs[%q] has no OrderBy and no SQL override (determinism contract)", sp.SheetName)
		}
	}
}

// ---------------------------------------------------------------------------
// composeOrgSheetSQL — drift-resistant SQL builder
// ---------------------------------------------------------------------------

func TestComposeOrgSheetSQL_AllColumnsPresent(t *testing.T) {
	spec := orgSheetSpec{
		SheetName: "appointments",
		Table:     "paliad.appointments",
		OrderBy:   []string{"id"},
	}
	cols := map[string]map[string]struct{}{
		"appointments": {"id": {}, "project_id": {}},
	}
	got, dropped := composeOrgSheetSQL(spec, cols)
	want := "SELECT * FROM paliad.appointments ORDER BY id"
	if got != want {
		t.Fatalf("got SQL %q, want %q", got, want)
	}
	if len(dropped) != 0 {
		t.Fatalf("expected no dropped columns, got %v", dropped)
	}
}

func TestComposeOrgSheetSQL_DropsMissingOrderByColumn(t *testing.T) {
	// The original bug from m/paliad#138 reproduced in unit form:
	// orderBy references a column the table doesn't have.
	spec := orgSheetSpec{
		SheetName: "appointment_caldav_targets",
		Table:     "paliad.appointment_caldav_targets",
		OrderBy:   []string{"appointment_id", "calendar_binding_id"}, // wrong: real col is binding_id
	}
	cols := map[string]map[string]struct{}{
		"appointment_caldav_targets": {
			"appointment_id": {},
			"binding_id":     {},
		},
	}
	got, dropped := composeOrgSheetSQL(spec, cols)
	want := "SELECT * FROM paliad.appointment_caldav_targets ORDER BY appointment_id"
	if got != want {
		t.Fatalf("got SQL %q, want %q", got, want)
	}
	if len(dropped) != 1 || dropped[0] != "calendar_binding_id" {
		t.Fatalf("expected dropped=[calendar_binding_id], got %v", dropped)
	}
}

func TestComposeOrgSheetSQL_AllOrderByMissing_NoClause(t *testing.T) {
	// If every declared ORDER BY column is gone, the builder still
	// produces a runnable SELECT — without ORDER BY. The export
	// succeeds; the order across runs is no longer deterministic for
	// this sheet until the spec is updated. WARN log alerts the
	// operator (verified in TestResolveOrgSheets_LogsWarnings).
	spec := orgSheetSpec{
		SheetName: "ghost",
		Table:     "paliad.ghost",
		OrderBy:   []string{"missing_a", "missing_b"},
	}
	cols := map[string]map[string]struct{}{
		"ghost": {"unrelated": {}},
	}
	got, dropped := composeOrgSheetSQL(spec, cols)
	want := "SELECT * FROM paliad.ghost"
	if got != want {
		t.Fatalf("got SQL %q, want %q", got, want)
	}
	if len(dropped) != 2 {
		t.Fatalf("expected 2 dropped columns, got %v", dropped)
	}
}

func TestComposeOrgSheetSQL_SQLOverride_BypassesBuilder(t *testing.T) {
	// When a sheet declares SQL, the builder MUST NOT touch it — even
	// if the column knowledge would suggest a change. Custom
	// projections (documents drops ai_extracted) and special-case
	// joins both rely on this.
	spec := orgSheetSpec{
		SheetName: "documents",
		Table:     "paliad.documents", // should be ignored
		OrderBy:   []string{"id"},     // should be ignored
		SQL:       "SELECT id, title FROM paliad.documents ORDER BY id",
	}
	cols := map[string]map[string]struct{}{
		"documents": {}, // empty → would drop everything if builder ran
	}
	got, dropped := composeOrgSheetSQL(spec, cols)
	if got != spec.SQL {
		t.Fatalf("SQL override mutated: got %q, want %q", got, spec.SQL)
	}
	if len(dropped) != 0 {
		t.Fatalf("override path should never report drops; got %v", dropped)
	}
}

func TestComposeOrgSheetSQL_UnknownTable_DropsAllOrderBy(t *testing.T) {
	// A table missing entirely from the schema snapshot is treated as
	// "no columns known" — every ORDER BY column gets dropped, but
	// the SELECT still emits (so a stale registry doesn't crash the
	// backup; the operator gets WARNs to fix it).
	spec := orgSheetSpec{
		SheetName: "renamed_table",
		Table:     "paliad.renamed_table",
		OrderBy:   []string{"id"},
	}
	got, dropped := composeOrgSheetSQL(spec, map[string]map[string]struct{}{})
	want := "SELECT * FROM paliad.renamed_table"
	if got != want {
		t.Fatalf("got SQL %q, want %q", got, want)
	}
	if len(dropped) != 1 || dropped[0] != "id" {
		t.Fatalf("expected dropped=[id], got %v", dropped)
	}
}

func TestComposeOrgSheetSQL_PreservesOrderByOrder(t *testing.T) {
	// Multi-column OrderBy must keep its declared order, with kept
	// columns concatenated in the same sequence. Determinism contract
	// from t-paliad-214 §3 depends on this.
	spec := orgSheetSpec{
		SheetName: "partner_unit_members",
		Table:     "paliad.partner_unit_members",
		OrderBy:   []string{"partner_unit_id", "missing_middle", "user_id"},
	}
	cols := map[string]map[string]struct{}{
		"partner_unit_members": {
			"partner_unit_id": {},
			"user_id":         {},
		},
	}
	got, dropped := composeOrgSheetSQL(spec, cols)
	want := "SELECT * FROM paliad.partner_unit_members ORDER BY partner_unit_id, user_id"
	if got != want {
		t.Fatalf("got SQL %q, want %q", got, want)
	}
	if len(dropped) != 1 || dropped[0] != "missing_middle" {
		t.Fatalf("expected dropped=[missing_middle], got %v", dropped)
	}
}

// ---------------------------------------------------------------------------
// LocalDiskStore round-trip
// ---------------------------------------------------------------------------

func TestLocalDiskStore_RoundTrip(t *testing.T) {
	dir := t.TempDir()
	store, err := NewLocalDiskStore(dir)
	if err != nil {
		t.Fatalf("NewLocalDiskStore: %v", err)
	}
	ctx := context.Background()
	want := []byte("hello backup\n")

	uri, err := store.Put(ctx, "test.zip", want)
	if err != nil {
		t.Fatalf("Put: %v", err)
	}
	if !strings.HasPrefix(uri, "file://") {
		t.Fatalf("expected file:// uri, got %q", uri)
	}
	rc, size, err := store.Get(ctx, uri)
	if err != nil {
		t.Fatalf("Get: %v", err)
	}
	defer rc.Close()
	if size != int64(len(want)) {
		t.Fatalf("Get size = %d, want %d", size, len(want))
	}
	got, err := io.ReadAll(rc)
	if err != nil {
		t.Fatalf("ReadAll: %v", err)
	}
	if !bytes.Equal(got, want) {
		t.Fatalf("Get body = %q, want %q", got, want)
	}
	if err := store.Delete(ctx, uri); err != nil {
		t.Fatalf("Delete: %v", err)
	}
	// File should be gone; Get returns an error.
	if _, _, err := store.Get(ctx, uri); err == nil {
		t.Fatalf("Get after Delete should fail")
	}
	// Delete is idempotent.
	if err := store.Delete(ctx, uri); err != nil {
		t.Fatalf("idempotent Delete: %v", err)
	}
}

func TestLocalDiskStore_RejectsBadKeys(t *testing.T) {
	dir := t.TempDir()
	store, err := NewLocalDiskStore(dir)
	if err != nil {
		t.Fatalf("NewLocalDiskStore: %v", err)
	}
	ctx := context.Background()
	cases := []string{
		"",
		"sub/dir/file.zip",
		"..\\evil.zip",
		"../escape.zip",
		"/abs/path.zip",
	}
	for _, k := range cases {
		if _, err := store.Put(ctx, k, []byte("x")); err == nil {
			t.Fatalf("Put with bad key %q should fail", k)
		}
	}
}

func TestLocalDiskStore_RejectsURIOutsideDir(t *testing.T) {
	dir := t.TempDir()
	store, err := NewLocalDiskStore(dir)
	if err != nil {
		t.Fatalf("NewLocalDiskStore: %v", err)
	}
	ctx := context.Background()
	// A file:// URI pointing outside the store dir must be rejected
	// by both Get and Delete (defense in depth against a corrupted
	// catalog row).
	outside := "file://" + filepath.Join(filepath.Dir(dir), "elsewhere.zip")
	if _, _, err := store.Get(ctx, outside); err == nil {
		t.Fatalf("Get outside store dir should fail")
	}
	if err := store.Delete(ctx, outside); err == nil {
		t.Fatalf("Delete outside store dir should fail")
	}
	// Wrong scheme is also rejected.
	if _, _, err := store.Get(ctx, "https://example.com/foo.zip"); err == nil {
		t.Fatalf("Get with non-file:// scheme should fail")
	}
}

func TestLocalDiskStore_CreatesDir(t *testing.T) {
	// A non-existent parent gets created at construction; mode 0700.
	base := t.TempDir()
	target := filepath.Join(base, "nested", "exports")
	store, err := NewLocalDiskStore(target)
	if err != nil {
		t.Fatalf("NewLocalDiskStore(non-existent): %v", err)
	}
	info, err := os.Stat(target)
	if err != nil {
		t.Fatalf("expected store dir to exist: %v", err)
	}
	if !info.IsDir() {
		t.Fatalf("expected directory, got file")
	}
	// Smoke-write to confirm the dir is actually usable.
	if _, err := store.Put(context.Background(), "ok.zip", []byte{}); err != nil {
		t.Fatalf("Put into fresh dir: %v", err)
	}
}