From ca6e271f665e1c1dd0112791fe376e4310f70b66 Mon Sep 17 00:00:00 2001 From: AJ ONeal Date: Wed, 16 Sep 2026 15:42:05 -0600 Subject: [PATCH] feat: add database/{csvx,rowsync,csvdb,pgdb,sheetsdb} --- database/csvdb/atomic.go | 45 ++++ database/csvdb/file.go | 304 +++++++++++++++++++++ database/csvdb/file_append_test.go | 229 ++++++++++++++++ database/csvdb/file_test.go | 145 ++++++++++ database/csvdb/gate.go | 34 +++ database/csvdb/test_types_test.go | 12 + database/csvx/accept_test.go | 37 +++ database/csvx/bool.go | 100 +++++++ database/csvx/boolish.go | 68 +++++ database/csvx/bytes.go | 108 ++++++++ database/csvx/csv.go | 76 ++++++ database/csvx/csvx_test.go | 229 ++++++++++++++++ database/csvx/date.go | 112 ++++++++ database/csvx/delimited.go | 175 ++++++++++++ database/csvx/delimited_test.go | 68 +++++ database/csvx/doc.go | 9 + database/csvx/encoding.go | 172 ++++++++++++ database/csvx/encoding_test.go | 42 +++ database/csvx/http.go | 130 +++++++++ database/csvx/normalize.go | 44 +++ database/csvx/number.go | 156 +++++++++++ database/csvx/rawjson.go | 42 +++ database/csvx/text.go | 89 +++++++ database/csvx/timestamp.go | 120 +++++++++ database/csvx/timestamptz.go | 111 ++++++++ database/csvx/uuid.go | 127 +++++++++ database/go.mod | 19 ++ database/go.sum | 38 +++ database/pgdb/conn.go | 172 ++++++++++++ database/rowsync/compose.go | 97 +++++++ database/rowsync/plan.go | 318 ++++++++++++++++++++++ database/rowsync/plan_test.go | 380 ++++++++++++++++++++++++++ database/rowsync/store.go | 52 ++++ database/sheetsdb/fetcher.go | 414 +++++++++++++++++++++++++++++ database/sheetsdb/fetcher_test.go | 21 ++ 35 files changed, 4295 insertions(+) create mode 100644 database/csvdb/atomic.go create mode 100644 database/csvdb/file.go create mode 100644 database/csvdb/file_append_test.go create mode 100644 database/csvdb/file_test.go create mode 100644 database/csvdb/gate.go create mode 100644 database/csvdb/test_types_test.go create mode 100644 database/csvx/accept_test.go create mode 100644 database/csvx/bool.go create mode 100644 database/csvx/boolish.go create mode 100644 database/csvx/bytes.go create mode 100644 database/csvx/csv.go create mode 100644 database/csvx/csvx_test.go create mode 100644 database/csvx/date.go create mode 100644 database/csvx/delimited.go create mode 100644 database/csvx/delimited_test.go create mode 100644 database/csvx/doc.go create mode 100644 database/csvx/encoding.go create mode 100644 database/csvx/encoding_test.go create mode 100644 database/csvx/http.go create mode 100644 database/csvx/normalize.go create mode 100644 database/csvx/number.go create mode 100644 database/csvx/rawjson.go create mode 100644 database/csvx/text.go create mode 100644 database/csvx/timestamp.go create mode 100644 database/csvx/timestamptz.go create mode 100644 database/csvx/uuid.go create mode 100644 database/go.mod create mode 100644 database/go.sum create mode 100644 database/pgdb/conn.go create mode 100644 database/rowsync/compose.go create mode 100644 database/rowsync/plan.go create mode 100644 database/rowsync/plan_test.go create mode 100644 database/rowsync/store.go create mode 100644 database/sheetsdb/fetcher.go create mode 100644 database/sheetsdb/fetcher_test.go diff --git a/database/csvdb/atomic.go b/database/csvdb/atomic.go new file mode 100644 index 0000000..89b8a2d --- /dev/null +++ b/database/csvdb/atomic.go @@ -0,0 +1,45 @@ +package csvdb + +import ( + "crypto/rand" + "encoding/base64" + "fmt" + "os" +) + +// createTemp creates a temporary file within root using the pattern +// ..tmp, where is a 32-bit random value encoded as +// URL-safe base64 (6 characters). The file is created with O_CREATE|O_EXCL +// and 0600 permissions. +// +// 2^32 possible names is sufficient to avoid collisions without retry. +func createTemp(root *os.Root, name string) (*os.File, string, error) { + var buf [4]byte + if _, err := rand.Read(buf[:]); err != nil { + return nil, "", fmt.Errorf("generate temp suffix: %w", err) + } + tmpName := name + "." + base64.RawURLEncoding.EncodeToString(buf[:]) + ".tmp" + f, err := root.OpenFile(tmpName, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0600) + if err != nil { + return nil, "", err + } + return f, tmpName, nil +} + +// AtomicWrite writes data to a temp file in root, then atomically renames it +// to name. If the write or rename fails, the temp file is removed. +func AtomicWrite(root *os.Root, name string, write func(*os.File) error) error { + tmp, tmpName, err := createTemp(root, name) + if err != nil { + return fmt.Errorf("create temp: %w", err) + } + defer func() { _ = root.Remove(tmpName) }() + if err := write(tmp); err != nil { + _ = tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("close temp: %w", err) + } + return root.Rename(tmpName, name) +} diff --git a/database/csvdb/file.go b/database/csvdb/file.go new file mode 100644 index 0000000..d99254f --- /dev/null +++ b/database/csvdb/file.go @@ -0,0 +1,304 @@ +// Package csvdb provides a generic CSV/TSV database adapter. +// Store implements both rowsync.Fetcher and rowsync.Updater over a file +// in an os.Root directory, with atomic writes and optional normalization. +package csvdb + +import ( + "bufio" + "bytes" + "context" + "encoding/csv" + "errors" + "io" + "os" + "path/filepath" + "strings" + + "github.com/therootcompany/golib/database/rowsync" + + "github.com/jszwec/csvutil" + + "github.com/therootcompany/golib/database/csvx" +) + +// Store is a generic TSV/CSV file storage adapter backed by an os.Root. +// Root must be non-nil; Name is the file within the root directory. +type Store[T any] struct { + Root *os.Root + Name string + Format string + Gate *WriteGate + Normalize func([]T) +} + +var _ rowsync.Fetcher[struct{}] = (*Store[struct{}])(nil) +var _ rowsync.Updater[struct{}] = (*Store[struct{}])(nil) + +// errHeaderMismatch signals that the existing file's header doesn't match +// the encoder's header. Update falls back to a full atomic rewrite. +var errHeaderMismatch = errors.New("tsv: existing file header does not match encoded header") + +// Init validates the root and file name. +// Combined fetches from several stores and concatenates their rows, e.g. +// clients.tsv + redirects.tsv, which share one row type and one target. +type Combined[T any] struct { + Stores []rowsync.Fetcher[T] +} + +func (c *Combined[T]) Init(_ context.Context) ([]rowsync.Message, error) { + if c == nil || len(c.Stores) == 0 { + return nil, errors.New("at least one store is required") + } + var messages []rowsync.Message + for _, s := range c.Stores { + msgs, err := s.Init(context.Background()) + if err != nil { + return nil, err + } + messages = append(messages, msgs...) + } + return messages, nil +} + +func (c *Combined[T]) Fetch(ctx context.Context) ([]T, error) { + if c == nil || len(c.Stores) == 0 { + return nil, errors.New("at least one store is required") + } + var rows []T + for _, s := range c.Stores { + part, err := s.Fetch(ctx) + if err != nil { + return nil, err + } + rows = append(rows, part...) + } + return rows, nil +} + +func (s *Store[T]) Init(_ context.Context) ([]rowsync.Message, error) { + if s == nil || s.Root == nil || strings.TrimSpace(s.Name) == "" { + return nil, errors.New("root and name are required") + } + return []rowsync.Message{{String: s.Name, Type: "debug"}}, nil +} + +// Fetch reads the existing file. Missing files return empty state. +func (s *Store[T]) Fetch(_ context.Context) ([]T, error) { + if s == nil || s.Root == nil || strings.TrimSpace(s.Name) == "" { + return nil, errors.New("root and name are required") + } + info, err := s.Root.Stat(s.Name) + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + if err != nil { + return nil, err + } + if info.Size() == 0 { + return nil, nil + } + file, err := s.Root.Open(s.Name) + if err != nil { + return nil, err + } + defer func() { _ = file.Close() }() + rows, err := parseFile[T](file, s.format()) + if err != nil { + return nil, err + } + if s.Normalize != nil { + s.Normalize(rows) + } + return rows, nil +} + +// Update applies a plan to the file. It serializes Plan.Final, never raw +// source rows: UpdateMode plans keep destination-only rows, SyncMode plans +// drop them. A plan with no actions leaves the file untouched. Insert-only +// plans append without rewriting the existing snapshot; any update or delete +// takes the atomic replacement path. The previous file survives encoding or +// rename failures (temp file + rename, no fsync). +func (s *Store[T]) Update(ctx context.Context, plan rowsync.Plan[T]) error { + if s == nil || s.Root == nil || strings.TrimSpace(s.Name) == "" { + return errors.New("root and name are required") + } + if err := ctx.Err(); err != nil { + return err + } + if len(plan.Actions) == 0 { + return nil + } + gate := s.Gate + if gate == nil { + gate = DefaultGate + } + if err := gate.acquire(ctx); err != nil { + return err + } + defer gate.release() + + // Insert-only plans can append without rewriting the existing snapshot. + // If the existing file's header doesn't match the encoder's header (e.g. + // the file was manually created from a template with different columns), + // fall through to the atomic replacement path so the whole file is + // rewritten with a consistent header. + inserts := make([]T, 0, len(plan.Actions)) + for _, action := range plan.Actions { + if action.Kind != rowsync.InsertAction { + break + } + inserts = append(inserts, *action.After) + } + if len(inserts) == len(plan.Actions) { + added := append([]T(nil), inserts...) + if s.Normalize != nil { + s.Normalize(added) + } + if err := s.appendFile(added); err == nil { + return nil + } + // Header mismatch or read error — fall through to full rewrite. + } + if plan.Final == nil { + return errors.New("plan final state is required for update or delete actions") + } + final := append([]T(nil), plan.Final...) + if s.Normalize != nil { + s.Normalize(final) + } + return AtomicWrite(s.Root, s.Name, func(tmp *os.File) error { + return writeFile[T](tmp, final, s.format()) + }) +} + +// WriteAll replaces the file with rows, unconditionally. Use it for +// full-state writes that are not driven by a plan (split files, fetched +// config snapshots). +func (s *Store[T]) WriteAll(ctx context.Context, rows []T) error { + if s == nil || s.Root == nil || strings.TrimSpace(s.Name) == "" { + return errors.New("root and name are required") + } + if err := ctx.Err(); err != nil { + return err + } + gate := s.Gate + if gate == nil { + gate = DefaultGate + } + if err := gate.acquire(ctx); err != nil { + return err + } + defer gate.release() + current := append([]T(nil), rows...) + if s.Normalize != nil { + s.Normalize(current) + } + return AtomicWrite(s.Root, s.Name, func(tmp *os.File) error { + return writeFile[T](tmp, current, s.format()) + }) +} + +func (s *Store[T]) appendFile(items []T) error { + if len(items) == 0 { + return nil + } + var encoded bytes.Buffer + if err := writeFile(&encoded, items, s.format()); err != nil { + return err + } + data := encoded.Bytes() + encodedHeader := data + if i := bytes.IndexByte(data, '\n'); i >= 0 { + encodedHeader = data[:i] + } + if info, err := s.Root.Stat(s.Name); err == nil && info.Size() > 0 { + existing, err := s.Root.Open(s.Name) + if err != nil { + return err + } + scanner := bufio.NewScanner(existing) + var existingHeader []byte + if scanner.Scan() { + existingHeader = scanner.Bytes() + } + _ = existing.Close() + if !bytes.Equal(existingHeader, encodedHeader) { + return errHeaderMismatch + } + newline := bytes.IndexByte(data, '\n') + if newline < 0 || newline+1 >= len(data) { + return errors.New("append TSV: encoded rows have no data record") + } + data = data[newline+1:] + } else if err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + file, err := s.Root.OpenFile(s.Name, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600) + if err != nil { + return err + } + defer func() { _ = file.Close() }() + writer := bufio.NewWriter(file) + if _, err := writer.Write(data); err != nil { + return err + } + return writer.Flush() +} + +func (s *Store[T]) format() string { + if strings.TrimSpace(s.Format) != "" { + return strings.ToLower(strings.TrimSpace(s.Format)) + } + return detectFormat(s.Name) +} + +func detectFormat(name string) string { + switch ext := strings.ToLower(filepath.Ext(name)); ext { + case ".tsv", ".tab": + return "tsv" + default: + return "csv" + } +} + +func parseFile[T any](r io.Reader, format string) ([]T, error) { + reader := csv.NewReader(r) + reader.Comment = '#' + if format == "tsv" { + reader.Comma = '\t' + } + reader.FieldsPerRecord = -1 + dec, err := csvutil.NewDecoder(reader) + if err != nil { + return nil, err + } + dec.Tag = "json" + dec.WithUnmarshalers(csvx.Unmarshalers()) + var items []T + for { + var item T + if err := dec.Decode(&item); err != nil { + if errors.Is(err, io.EOF) { + break + } + return nil, err + } + items = append(items, item) + } + return items, nil +} + +func writeFile[T any](w io.Writer, items []T, format string) error { + cw := csv.NewWriter(w) + if format == "tsv" { + cw.Comma = '\t' + } + enc := csvutil.NewEncoder(cw) + enc.Tag = "json" + enc.WithMarshalers(csvx.Marshalers()) + if err := enc.Encode(items); err != nil { + return err + } + cw.Flush() + return cw.Error() +} diff --git a/database/csvdb/file_append_test.go b/database/csvdb/file_append_test.go new file mode 100644 index 0000000..5e6e19b --- /dev/null +++ b/database/csvdb/file_append_test.go @@ -0,0 +1,229 @@ +package csvdb + +import ( + "context" + "os" + "testing" + + "github.com/therootcompany/golib/database/rowsync" +) + +func insertPlan[T any](key string, rows ...T) rowsync.Plan[T] { + plan := rowsync.Plan[T]{Mode: rowsync.UpdateMode} + for i := range rows { + row := rows[i] + plan.Actions = append(plan.Actions, rowsync.Action[T]{Kind: rowsync.InsertAction, Key: key, After: &row}) + plan.Final = append(plan.Final, row) + } + return plan +} + +func TestFileAppendFastPathPreservesPrefix(t *testing.T) { + dir := t.TempDir() + root, err := os.OpenRoot(dir) + if err != nil { + t.Fatal(err) + } + defer root.Close() + store := &Store[testRuleRow]{Root: root, Name: "rules.tsv", Format: "tsv"} + ctx := context.Background() + if err := store.WriteAll(ctx, []testRuleRow{{Email: "a@example.test"}}); err != nil { + t.Fatal(err) + } + before, _ := os.ReadFile(dir + "/rules.tsv") + plan := insertPlan[testRuleRow]("b@example.test", testRuleRow{Email: "b@example.test"}) + if err := store.Update(ctx, plan); err != nil { + t.Fatal(err) + } + after, _ := os.ReadFile(dir + "/rules.tsv") + if string(after[:len(before)]) != string(before) { + t.Fatal("append fast path changed existing bytes") + } + rows, err := store.Fetch(ctx) + if err != nil || len(rows) != 2 { + t.Fatalf("rows=%d err=%v", len(rows), err) + } +} + +func TestFileAppendHeaderMismatchFallsBackToRewrite(t *testing.T) { + dir := t.TempDir() + root, err := os.OpenRoot(dir) + if err != nil { + t.Fatal(err) + } + defer root.Close() + store := &Store[testRuleRow]{Root: root, Name: "rules.tsv", Format: "tsv"} + ctx := context.Background() + + // Create a file with a manually-written header that has fewer columns + // than the struct produces. This simulates the production bug: a + // template-derived TSV whose header doesn't match the encoder output. + if err := root.WriteFile("rules.tsv", []byte("email\nold@example.test\n"), 0600); err != nil { + t.Fatal(err) + } + + // Insert-only plan: Update must detect the header mismatch and fall + // back to a full atomic rewrite instead of appending mismatched rows. + newRow := testRuleRow{Email: "new@example.test"} + plan := insertPlan[testRuleRow]("new@example.test", newRow) + plan.Final = []testRuleRow{ + {Email: "old@example.test"}, + newRow, + } + if err := store.Update(ctx, plan); err != nil { + t.Fatalf("Update with header mismatch: %v", err) + } + + rows, err := store.Fetch(ctx) + if err != nil { + t.Fatalf("Fetch after rewrite: %v", err) + } + if len(rows) != 2 { + t.Fatalf("got %d rows, want 2", len(rows)) + } +} + +func TestFileUpdateFallsBackToRewrite(t *testing.T) { + dir := t.TempDir() + root, err := os.OpenRoot(dir) + if err != nil { + t.Fatal(err) + } + defer root.Close() + store := &Store[testRuleRow]{Root: root, Name: "rules.tsv", Format: "tsv"} + ctx := context.Background() + initial := []testRuleRow{{Email: "a@example.test"}} + if err := store.WriteAll(ctx, initial); err != nil { + t.Fatal(err) + } + changed := testRuleRow{Email: "changed@example.test"} + plan := rowsync.Plan[testRuleRow]{ + Mode: rowsync.SyncMode, + Final: []testRuleRow{changed}, + Actions: []rowsync.Action[testRuleRow]{{ + Kind: rowsync.UpdateAction, + Key: "changed@example.test", + Before: &initial[0], + After: &changed, + }}, + } + if err := store.Update(ctx, plan); err != nil { + t.Fatal(err) + } + rows, err := store.Fetch(ctx) + if err != nil || len(rows) != 1 || rows[0].Email != "changed@example.test" { + t.Fatalf("rows=%#v err=%v", rows, err) + } +} + +func TestFileUpdateRequiresFinalState(t *testing.T) { + dir := t.TempDir() + root, err := os.OpenRoot(dir) + if err != nil { + t.Fatal(err) + } + defer root.Close() + store := &Store[testRuleRow]{Root: root, Name: "rules.tsv", Format: "tsv"} + ctx := context.Background() + if err := store.WriteAll(ctx, []testRuleRow{{Email: "a@example.test"}}); err != nil { + t.Fatal(err) + } + before, _ := os.ReadFile(dir + "/rules.tsv") + row := testRuleRow{Email: "a@example.test"} + plan := rowsync.Plan[testRuleRow]{ + Actions: []rowsync.Action[testRuleRow]{{Kind: rowsync.DeleteAction, Key: "a@example.test", Before: &row}}, + } + if err := store.Update(ctx, plan); err == nil { + t.Fatal("Update accepted a plan without a final state") + } + after, _ := os.ReadFile(dir + "/rules.tsv") + if string(before) != string(after) { + t.Fatal("failed Update modified the file") + } +} + +func TestFileFailedWritePreservesPreviousFile(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("running as root; permission bits do not apply") + } + dir := t.TempDir() + root, err := os.OpenRoot(dir) + if err != nil { + t.Fatal(err) + } + defer root.Close() + store := &Store[testRuleRow]{Root: root, Name: "rules.tsv", Format: "tsv"} + ctx := context.Background() + original := []testRuleRow{{Email: "a@example.test"}} + if err := store.WriteAll(ctx, original); err != nil { + t.Fatal(err) + } + if err := os.Chmod(dir, 0555); err != nil { + t.Fatal(err) + } + defer func() { _ = os.Chmod(dir, 0755) }() + row := testRuleRow{Email: "changed@example.test"} + plan := rowsync.Plan[testRuleRow]{ + Mode: rowsync.SyncMode, + Final: []testRuleRow{row}, + Actions: []rowsync.Action[testRuleRow]{{ + Kind: rowsync.UpdateAction, + Key: "changed@example.test", + Before: &original[0], + After: &row, + }}, + } + if err := store.Update(ctx, plan); err == nil { + t.Fatal("Update succeeded on a read-only directory") + } + rows, err := store.Fetch(ctx) + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 || rows[0].Email != "a@example.test" { + t.Fatalf("previous file not preserved: %#v", rows) + } + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + if len(entries) != 1 || entries[0].Name() != "rules.tsv" { + t.Fatalf("leftover temp files: %v", entries) + } +} + +func TestFileZeroActionPlanLeavesFileUnchanged(t *testing.T) { + dir := t.TempDir() + root, err := os.OpenRoot(dir) + if err != nil { + t.Fatal(err) + } + defer root.Close() + store := &Store[testRuleRow]{Root: root, Name: "rules.tsv", Format: "tsv"} + ctx := context.Background() + if err := store.WriteAll(ctx, []testRuleRow{{Email: "a@example.test"}}); err != nil { + t.Fatal(err) + } + before, err := os.ReadFile(dir + "/rules.tsv") + if err != nil { + t.Fatal(err) + } + statBefore, err := os.Stat(dir + "/rules.tsv") + if err != nil { + t.Fatal(err) + } + if err := store.Update(ctx, rowsync.Plan[testRuleRow]{Mode: rowsync.UpdateMode, Final: []testRuleRow{{Email: "a@example.test"}}}); err != nil { + t.Fatal(err) + } + after, err := os.ReadFile(dir + "/rules.tsv") + if err != nil { + t.Fatal(err) + } + statAfter, err := os.Stat(dir + "/rules.tsv") + if err != nil { + t.Fatal(err) + } + if string(before) != string(after) || !statBefore.ModTime().Equal(statAfter.ModTime()) { + t.Fatal("zero-action plan rewrote the file") + } +} diff --git a/database/csvdb/file_test.go b/database/csvdb/file_test.go new file mode 100644 index 0000000..b3cfdc5 --- /dev/null +++ b/database/csvdb/file_test.go @@ -0,0 +1,145 @@ +package csvdb + +import ( + "context" + "os" + "testing" + + "github.com/therootcompany/golib/database/csvx" +) + +func TestFileRoundtrip(t *testing.T) { + root, err := os.OpenRoot(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = root.Close() }() + f := &Store[testRuleRow]{Root: root, Name: "test.tsv", Format: "tsv"} + ctx := context.Background() + + if _, err := f.Init(ctx); err != nil { + t.Fatal(err) + } + rows := []testRuleRow{ + {Email: "a@example.com"}, + {Email: "b@example.com"}, + } + if err := f.WriteAll(ctx, rows); err != nil { + t.Fatal(err) + } + got, err := f.Fetch(ctx) + if err != nil { + t.Fatal(err) + } + if len(got) != 2 { + t.Fatalf("got %d rows, want 2", len(got)) + } + if got[0].Email != "a@example.com" || got[1].Email != "b@example.com" { + t.Fatalf("unexpected rows: %v", got) + } +} + +func TestFileNormalizeHook(t *testing.T) { + root, err := os.OpenRoot(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = root.Close() }() + f := &Store[identityRow]{ + Root: root, + Name: "identities.tsv", + Format: "tsv", + Normalize: func(rows []identityRow) { + for i := range rows { + if rows[i].Email.Valid { + rows[i].Email.String = csvx.NormalizeEmail(rows[i].Email.String) + } + if rows[i].Phone.Valid { + rows[i].Phone.String = csvx.NormalizePhone(rows[i].Phone.String) + } + } + }, + } + ctx := context.Background() + rows := []identityRow{{ + Email: csvx.Text{String: " USER@Example.COM ", Valid: true}, + Phone: csvx.Text{String: "+1 (415) 555-0123", Valid: true}, + }} + if err := f.WriteAll(ctx, rows); err != nil { + t.Fatal(err) + } + got, err := f.Fetch(ctx) + if err != nil { + t.Fatal(err) + } + if len(got) != 1 || got[0].Email.String != "user@example.com" || got[0].Phone.String != "(415) 555-0123" { + t.Fatalf("contacts = %#v", got) + } +} + +func TestFileFetchSkipsComments(t *testing.T) { + root, err := os.OpenRoot(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = root.Close() }() + content := "# beginner note\nemail\trevoked_at\texpires_at\nuser@example.com\t\t\n" + if err := root.WriteFile("comments.tsv", []byte(content), 0600); err != nil { + t.Fatal(err) + } + f := &Store[testRuleRow]{Root: root, Name: "comments.tsv", Format: "tsv"} + got, err := f.Fetch(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(got) != 1 || got[0].Email != "user@example.com" { + t.Fatalf("got %#v, want one parsed row", got) + } +} + +func TestFileFetchMissing(t *testing.T) { + root, err := os.OpenRoot(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = root.Close() }() + f := &Store[testRuleRow]{Root: root, Name: "missing.tsv", Format: "tsv"} + got, err := f.Fetch(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(got) != 0 { + t.Fatalf("got %d rows, want 0", len(got)) + } +} + +func TestFileAtomicWrite(t *testing.T) { + dir := t.TempDir() + root, err := os.OpenRoot(dir) + if err != nil { + t.Fatal(err) + } + defer func() { _ = root.Close() }() + f := &Store[testRuleRow]{Root: root, Name: "test.tsv", Format: "tsv"} + ctx := context.Background() + if _, err := f.Init(ctx); err != nil { + t.Fatal(err) + } + rows := []testRuleRow{{Email: "x@example.com"}} + for range 5 { + if err := f.WriteAll(ctx, rows); err != nil { + t.Fatal(err) + } + } + info, err := root.Stat("test.tsv") + if err != nil { + t.Fatal(err) + } + if info.Size() == 0 { + t.Fatal("file is empty after writes") + } + entries, _ := os.ReadDir(dir) + if len(entries) != 1 { + t.Fatalf("expected 1 file in dir, got %d (temp files not cleaned up?)", len(entries)) + } +} diff --git a/database/csvdb/gate.go b/database/csvdb/gate.go new file mode 100644 index 0000000..c55e8fd --- /dev/null +++ b/database/csvdb/gate.go @@ -0,0 +1,34 @@ +package csvdb + +import "context" + +// WriteGate permits one local file write at a time. +type WriteGate struct { + slots chan struct{} +} + +// NewWriteGate creates a gate that permits one write at a time. +func NewWriteGate() *WriteGate { + return &WriteGate{slots: make(chan struct{}, 1)} +} + +func (g *WriteGate) acquire(ctx context.Context) error { + if g == nil { + return nil + } + select { + case g.slots <- struct{}{}: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (g *WriteGate) release() { + if g != nil { + <-g.slots + } +} + +// DefaultGate is the shared write gate for the process. +var DefaultGate = NewWriteGate() diff --git a/database/csvdb/test_types_test.go b/database/csvdb/test_types_test.go new file mode 100644 index 0000000..14e6ad7 --- /dev/null +++ b/database/csvdb/test_types_test.go @@ -0,0 +1,12 @@ +package csvdb + +import "github.com/therootcompany/golib/database/csvx" + +type testRuleRow struct { + Email string `json:"email"` +} + +type identityRow struct { + Email csvx.Text `json:"email"` + Phone csvx.Text `json:"phone"` +} diff --git a/database/csvx/accept_test.go b/database/csvx/accept_test.go new file mode 100644 index 0000000..bdf6937 --- /dev/null +++ b/database/csvx/accept_test.go @@ -0,0 +1,37 @@ +package csvx + +import ( + "net/http/httptest" + "testing" +) + +func TestAcceptFormat(t *testing.T) { + for _, tc := range []struct { + accept, want string + }{ + {"application/json", "json"}, + {"text/csv", "csv"}, + {"text/tab-separated-values", "tab"}, + {"application/x-ndjson", "jsonl"}, + {"*/*", "csv"}, + {"", "csv"}, + {"text/html", "csv"}, + {"application/json;q=0.5, text/csv", "csv"}, + {"text/csv;q=0.5, application/json", "json"}, + } { + r := httptest.NewRequest("GET", "/x", nil) + r.Header.Set("Accept", tc.accept) + got, err := ResolveFormat(r) + if err != nil || string(got) != tc.want { + t.Errorf("Accept=%q → %q (err %v), want %q", tc.accept, got, err, tc.want) + } + } +} + +func TestAcceptFormatDirect(t *testing.T) { + got, err := acceptFormat("application/json") + t.Logf("acceptFormat(application/json) = %q, %v", got, err) + if string(got) != "json" { + t.Fatalf("want json, got %q", got) + } +} diff --git a/database/csvx/bool.go b/database/csvx/bool.go new file mode 100644 index 0000000..0f84937 --- /dev/null +++ b/database/csvx/bool.go @@ -0,0 +1,100 @@ +package csvx + +import ( + "database/sql/driver" + "encoding/json" + "fmt" + "strconv" + "strings" +) + +// Bool is a nullable boolean that implements CSV, JSON, and driver interfaces. +type Bool struct { + Bool bool + Valid bool +} + +func (b Bool) IsNil() bool { return !b.Valid } + +func (b Bool) Value() (driver.Value, error) { + if !b.Valid { + return nil, nil + } + return b.Bool, nil +} + +func (b *Bool) Scan(value any) error { + if value == nil { + b.Valid = false + b.Bool = false + return nil + } + switch v := value.(type) { + case bool: + b.Bool = v + b.Valid = true + case string: + parsed, err := strconv.ParseBool(strings.TrimSpace(v)) + if err != nil { + b.Valid = false + b.Bool = false + return nil + } + b.Bool = parsed + b.Valid = true + case []byte: + return b.Scan(string(v)) + case Bool: + *b = v + default: + return fmt.Errorf("csvx.Bool: cannot scan %T", value) + } + return nil +} + +func (b Bool) MarshalCSV() ([]byte, error) { + if !b.Valid { + return nil, nil + } + return []byte(strconv.FormatBool(b.Bool)), nil +} + +func (b *Bool) UnmarshalCSV(data []byte) error { + s := strings.TrimSpace(string(data)) + if s == "" { + b.Valid = false + b.Bool = false + return nil + } + parsed, err := strconv.ParseBool(s) + if err != nil { + b.Valid = false + b.Bool = false + return nil + } + b.Bool = parsed + b.Valid = true + return nil +} + +func (b Bool) MarshalJSON() ([]byte, error) { + if !b.Valid { + return []byte("null"), nil + } + return json.Marshal(b.Bool) +} + +func (b *Bool) UnmarshalJSON(data []byte) error { + if string(data) == "null" { + b.Valid = false + b.Bool = false + return nil + } + var v bool + if err := json.Unmarshal(data, &v); err != nil { + return err + } + b.Bool = v + b.Valid = true + return nil +} diff --git a/database/csvx/boolish.go b/database/csvx/boolish.go new file mode 100644 index 0000000..d5951af --- /dev/null +++ b/database/csvx/boolish.go @@ -0,0 +1,68 @@ +package csvx + +import ( + "fmt" + "strings" + "time" +) + +// BoolishTimestamp accepts optional true/false-like values and dates or +// timestamps from human-edited CSV/TSV data. Valid reports that a value was +// supplied; Bool reports its true-like meaning. Error records an unrecognized +// non-empty value without making the CSV decoder reject the row. +type BoolishTimestamp struct { + Valid bool + Bool bool + Timestamp time.Time + Error error +} + +// UnmarshalCSV parses a human value. Unknown non-empty values are retained as +// errors so the collection mapper can apply its own conservative policy. +func (v *BoolishTimestamp) UnmarshalCSV(data []byte) error { + *v = BoolishTimestamp{} + raw := strings.TrimSpace(string(data)) + if raw == "" || strings.EqualFold(raw, "null") || raw == `\N` { + return nil + } + v.Valid = true + switch strings.ToLower(raw) { + case "false", "f", "n", "no", "0": + return nil + case "true", "t", "y", "yes", "1", "disabled", "disable": + v.Bool = true + return nil + } + for _, layout := range []string{time.RFC3339, "2006-01-02", "2006-01-02 15:04:05Z07:00", "2006-01-02 15:04:05"} { + parsed, err := time.Parse(layout, raw) + if err == nil { + v.Bool = true + v.Timestamp = parsed + return nil + } + } + v.Bool = true + v.Error = fmt.Errorf("unrecognized boolish timestamp %q", raw) + return nil +} + +// MarshalCSV writes a normalized usable record. An unrecognized value is +// conservatively serialized as the current UTC timestamp. +func (v BoolishTimestamp) MarshalCSV() ([]byte, error) { + if !v.Valid { + return nil, nil + } + if v.Error != nil { + return []byte(time.Now().UTC().Format("2006-01-02 15:04:05")), nil + } + if !v.Bool { + return []byte("FALSE"), nil + } + if !v.Timestamp.IsZero() { + return []byte(v.Timestamp.UTC().Format("2006-01-02 15:04:05")), nil + } + return []byte("TRUE"), nil +} + +// IsZero lets encoders treat the zero value as an omitted source value. +func (v BoolishTimestamp) IsZero() bool { return !v.Valid } diff --git a/database/csvx/bytes.go b/database/csvx/bytes.go new file mode 100644 index 0000000..b0d9e4d --- /dev/null +++ b/database/csvx/bytes.go @@ -0,0 +1,108 @@ +package csvx + +import ( + "encoding/base64" + "encoding/hex" + "encoding/json" + "fmt" +) + +// URLBase64 unmarshals to bytes and marshals to a raw url base64 string +type URLBase64 []byte + +func (s URLBase64) String() string { + encoded := base64.RawURLEncoding.EncodeToString(s) + return encoded +} + +func (s URLBase64) MarshalCSV() ([]byte, error) { + return []byte(s.String()), nil +} + +func (s *URLBase64) UnmarshalCSV(src []byte) error { + decoded, err := base64.RawURLEncoding.DecodeString(string(src)) + if err != nil { + return fmt.Errorf("decode base64url: %w", err) + } + *s = decoded + return nil +} + +// MarshalJSON implements JSON marshaling to URL-safe base64. +func (s URLBase64) MarshalJSON() ([]byte, error) { + encoded := base64.RawURLEncoding.EncodeToString(s) + return json.Marshal(encoded) +} + +// UnmarshalJSON implements JSON unmarshaling from URL-safe base64. +func (s *URLBase64) UnmarshalJSON(data []byte) error { + dst, err := base64.RawURLEncoding.AppendDecode([]byte{}, data) + if err != nil { + return fmt.Errorf("decode base64url signature: %w", err) + } + + *s = dst + return nil +} + +// RFCBase64 unmarshals to bytes and marshals to standard base64. +type RFCBase64 []byte + +func (s RFCBase64) String() string { + encoded := base64.StdEncoding.EncodeToString(s) + return encoded +} + +// MarshalJSON implements JSON marshaling to standard base64. +func (s RFCBase64) MarshalJSON() ([]byte, error) { + encoded := base64.StdEncoding.EncodeToString(s) + return json.Marshal(encoded) +} + +// UnmarshalJSON implements JSON unmarshaling from standard base64. +func (s *RFCBase64) UnmarshalJSON(data []byte) error { + dst, err := base64.StdEncoding.AppendDecode([]byte{}, data) + if err != nil { + return fmt.Errorf("decode base64 signature: %w", err) + } + + *s = dst + return nil +} + +// Hex marshals and unmarshals bytes to a base16 (hex) string as JSON or CSV. +type Hex []byte + +func (h Hex) MarshalJSON() ([]byte, error) { + if h == nil { + return json.Marshal(nil) + } + encoded := hex.EncodeToString(h) + return json.Marshal(encoded) +} + +func (h *Hex) UnmarshalJSON(src []byte) error { + var s string + if err := json.Unmarshal(src, &s); err != nil { + return err + } + decoded, err := hex.DecodeString(s) + if err != nil { + return fmt.Errorf("decode hex: %w", err) + } + *h = decoded + return nil +} + +func (h Hex) MarshalCSV() ([]byte, error) { + return []byte(hex.EncodeToString(h)), nil +} + +func (h *Hex) UnmarshalCSV(src []byte) error { + decoded, err := hex.DecodeString(string(src)) + if err != nil { + return fmt.Errorf("decode hex: %w", err) + } + *h = decoded + return nil +} diff --git a/database/csvx/csv.go b/database/csvx/csv.go new file mode 100644 index 0000000..fd5f5bc --- /dev/null +++ b/database/csvx/csv.go @@ -0,0 +1,76 @@ +package csvx + +import ( + "encoding/hex" + "strconv" + "strings" + + "github.com/jszwec/csvutil" +) + +// Marshalers returns csvutil.Marshalers for generic types that don't have +// their own MarshalCSV method. +func Marshalers() *csvutil.Marshalers { + return csvutil.NewMarshalers( + csvutil.MarshalFunc[[]string](func(s []string) ([]byte, error) { + if len(s) == 0 { + return nil, nil + } + return []byte(strings.Join(s, ",")), nil + }), + csvutil.MarshalFunc[*bool](func(b *bool) ([]byte, error) { + if b == nil { + return nil, nil + } + return []byte(strconv.FormatBool(*b)), nil + }), + csvutil.MarshalFunc[[]byte](func(b []byte) ([]byte, error) { + if len(b) == 0 { + return nil, nil + } + return []byte(hex.EncodeToString(b)), nil + }), + ) +} + +// Unmarshalers returns csvutil.Unmarshalers for generic types that don't have +// their own UnmarshalCSV method. +func Unmarshalers() *csvutil.Unmarshalers { + return csvutil.NewUnmarshalers( + csvutil.UnmarshalFunc[*[]string](func(data []byte, s *[]string) error { + str := string(data) + if str == "" { + *s = nil + return nil + } + *s = strings.Split(str, ",") + return nil + }), + csvutil.UnmarshalFunc[*bool](func(data []byte, b *bool) error { + s := string(data) + if s == "" { + *b = false + return nil + } + v, err := strconv.ParseBool(strings.TrimSpace(s)) + if err != nil { + return err + } + *b = v + return nil + }), + csvutil.UnmarshalFunc[*[]byte](func(data []byte, b *[]byte) error { + s := string(data) + if s == "" { + *b = nil + return nil + } + decoded, err := hex.DecodeString(s) + if err != nil { + return err + } + *b = decoded + return nil + }), + ) +} diff --git a/database/csvx/csvx_test.go b/database/csvx/csvx_test.go new file mode 100644 index 0000000..ed45be5 --- /dev/null +++ b/database/csvx/csvx_test.go @@ -0,0 +1,229 @@ +package csvx + +import ( + "testing" + "time" + + "github.com/google/uuid" +) + +func TestUUIDRoundtrip(t *testing.T) { + // Valid UUID + u := UUID{Bytes: [16]byte(uuid.New()), Valid: true} + + // CSV roundtrip + data, err := u.MarshalCSV() + if err != nil { + t.Fatalf("MarshalCSV: %v", err) + } + var u2 UUID + if err := u2.UnmarshalCSV(data); err != nil { + t.Fatalf("UnmarshalCSV: %v", err) + } + if u2.String() != u.String() { + t.Fatalf("CSV roundtrip mismatch: %s != %s", u2.String(), u.String()) + } + + // Nil UUID + var nilU UUID + data, err = nilU.MarshalCSV() + if err != nil { + t.Fatalf("MarshalCSV nil: %v", err) + } + if data != nil { + t.Fatalf("expected nil for invalid UUID, got %v", data) + } + var nilU2 UUID + if err := nilU2.UnmarshalCSV(nil); err != nil { + t.Fatalf("UnmarshalCSV nil: %v", err) + } + if nilU2.Valid { + t.Fatal("expected invalid after nil unmarshal") + } + + // driver.Value roundtrip + v, err := u.Value() + if err != nil { + t.Fatalf("Value: %v", err) + } + if s, ok := v.(string); !ok || s != u.String() { + t.Fatalf("Value returned %T %v, want string %s", v, v, u.String()) + } + + // Scan from string + var scanned UUID + if err := scanned.Scan(u.String()); err != nil { + t.Fatalf("Scan string: %v", err) + } + if scanned.String() != u.String() { + t.Fatalf("Scan mismatch: %s != %s", scanned.String(), u.String()) + } + + // Scan from nil + if err := scanned.Scan(nil); err != nil { + t.Fatalf("Scan nil: %v", err) + } + if scanned.Valid { + t.Fatal("expected invalid after Scan(nil)") + } +} + +func TestTextRoundtrip(t *testing.T) { + tt := Text{String: "hello", Valid: true} + + data, err := tt.MarshalCSV() + if err != nil { + t.Fatalf("MarshalCSV: %v", err) + } + if string(data) != "hello" { + t.Fatalf("MarshalCSV: got %q", data) + } + var tt2 Text + if err := tt2.UnmarshalCSV(data); err != nil { + t.Fatalf("UnmarshalCSV: %v", err) + } + if tt2.String != "hello" || !tt2.Valid { + t.Fatalf("UnmarshalCSV: got %+v", tt2) + } + + // Nil + var nilT Text + data, _ = nilT.MarshalCSV() + if data != nil { + t.Fatalf("expected nil, got %v", data) + } + + // driver.Value + v, _ := tt.Value() + if v != "hello" { + t.Fatalf("Value: got %v", v) + } +} + +func TestTimestampRoundtrip(t *testing.T) { + now := time.Date(2025, 1, 15, 10, 30, 0, 0, time.UTC) + ts := Timestamp{Time: now, Valid: true} + + data, err := ts.MarshalCSV() + if err != nil { + t.Fatalf("MarshalCSV: %v", err) + } + var ts2 Timestamp + if err := ts2.UnmarshalCSV(data); err != nil { + t.Fatalf("UnmarshalCSV: %v", err) + } + if !ts2.Time.Equal(now) { + t.Fatalf("roundtrip: %s != %s", ts2.Time, now) + } + + // driver.Value returns time.Time + v, _ := ts.Value() + if v != now { + t.Fatalf("Value: got %T %v", v, v) + } + + // Scan from time.Time + var scanned Timestamp + if err := scanned.Scan(now); err != nil { + t.Fatalf("Scan time.Time: %v", err) + } + if !scanned.Time.Equal(now) || !scanned.Valid { + t.Fatalf("Scan: got %+v", scanned) + } +} + +func TestTimestamptzRoundtrip(t *testing.T) { + now := time.Date(2025, 6, 20, 14, 0, 0, 0, time.UTC) + tz := Timestamptz{Time: now, Valid: true} + + data, err := tz.MarshalCSV() + if err != nil { + t.Fatalf("MarshalCSV: %v", err) + } + var tz2 Timestamptz + if err := tz2.UnmarshalCSV(data); err != nil { + t.Fatalf("UnmarshalCSV: %v", err) + } + if !tz2.Time.Equal(now) { + t.Fatalf("roundtrip: %s != %s", tz2.Time, now) + } +} + +func TestDateRoundtrip(t *testing.T) { + d := Date{Time: time.Date(2025, 3, 15, 0, 0, 0, 0, time.UTC), Valid: true} + + data, err := d.MarshalCSV() + if err != nil { + t.Fatalf("MarshalCSV: %v", err) + } + if string(data) != "2025-03-15" { + t.Fatalf("MarshalCSV: got %q", data) + } + var d2 Date + if err := d2.UnmarshalCSV(data); err != nil { + t.Fatalf("UnmarshalCSV: %v", err) + } + if d2.Time.Day() != 15 || d2.Time.Month() != 3 || d2.Time.Year() != 2025 { + t.Fatalf("UnmarshalCSV: got %s", d2.Time) + } +} + +func TestBoolRoundtrip(t *testing.T) { + b := Bool{Bool: true, Valid: true} + + data, err := b.MarshalCSV() + if err != nil { + t.Fatalf("MarshalCSV: %v", err) + } + if string(data) != "true" { + t.Fatalf("MarshalCSV: got %q", data) + } + var b2 Bool + if err := b2.UnmarshalCSV(data); err != nil { + t.Fatalf("UnmarshalCSV: %v", err) + } + if !b2.Bool || !b2.Valid { + t.Fatalf("UnmarshalCSV: got %+v", b2) + } + + // driver.Value + v, _ := b.Value() + if v != true { + t.Fatalf("Value: got %v", v) + } +} + +func TestBoolishTimestamp(t *testing.T) { + // True-like + var v BoolishTimestamp + if err := v.UnmarshalCSV([]byte("true")); err != nil { + t.Fatalf("UnmarshalCSV: %v", err) + } + if !v.Valid || !v.Bool { + t.Fatalf("expected valid+true, got %+v", v) + } + + // False-like + if err := v.UnmarshalCSV([]byte("no")); err != nil { + t.Fatalf("UnmarshalCSV: %v", err) + } + if !v.Valid || v.Bool { + t.Fatalf("expected valid+false, got %+v", v) + } + + // Empty + if err := v.UnmarshalCSV([]byte("")); err != nil { + t.Fatalf("UnmarshalCSV: %v", err) + } + if v.Valid { + t.Fatal("expected invalid for empty") + } + + // Timestamp + if err := v.UnmarshalCSV([]byte("2025-01-15")); err != nil { + t.Fatalf("UnmarshalCSV: %v", err) + } + if !v.Valid || !v.Bool || v.Timestamp.IsZero() { + t.Fatalf("expected valid+true+timestamp, got %+v", v) + } +} diff --git a/database/csvx/date.go b/database/csvx/date.go new file mode 100644 index 0000000..d22450a --- /dev/null +++ b/database/csvx/date.go @@ -0,0 +1,112 @@ +package csvx + +import ( + "database/sql/driver" + "encoding/json" + "fmt" + "time" +) + +// Date is a nullable date that implements CSV, JSON, and driver interfaces. +type Date struct { + Time time.Time + Valid bool +} + +const dateFormat = "2006-01-02" + +func (d Date) IsNil() bool { return !d.Valid } + +func (d Date) Value() (driver.Value, error) { + if !d.Valid { + return nil, nil + } + return d.Time, nil +} + +func (d *Date) Scan(value any) error { + if value == nil { + d.Valid = false + d.Time = time.Time{} + return nil + } + switch v := value.(type) { + case time.Time: + d.Time = v + d.Valid = true + case string: + parsed, err := time.Parse(dateFormat, v) + if err != nil { + d.Valid = false + d.Time = time.Time{} + return nil + } + d.Time = parsed + d.Valid = true + case []byte: + return d.Scan(string(v)) + case Date: + *d = v + default: + return fmt.Errorf("csvx.Date: cannot scan %T", value) + } + return nil +} + +func (d Date) MarshalCSV() ([]byte, error) { + if !d.Valid { + return nil, nil + } + return []byte(d.Time.Format(dateFormat)), nil +} + +func (d *Date) UnmarshalCSV(data []byte) error { + s := string(data) + if s == "" { + d.Valid = false + d.Time = time.Time{} + return nil + } + parsed, err := time.Parse(dateFormat, s) + if err != nil { + d.Valid = false + d.Time = time.Time{} + return nil + } + d.Time = parsed + d.Valid = true + return nil +} + +func (d Date) MarshalJSON() ([]byte, error) { + if !d.Valid { + return []byte("null"), nil + } + return json.Marshal(d.Time.Format(dateFormat)) +} + +func (d *Date) UnmarshalJSON(data []byte) error { + if string(data) == "null" { + d.Valid = false + d.Time = time.Time{} + return nil + } + var s string + if err := json.Unmarshal(data, &s); err != nil { + return err + } + if s == "" { + d.Valid = false + d.Time = time.Time{} + return nil + } + parsed, err := time.Parse(dateFormat, s) + if err != nil { + d.Valid = false + d.Time = time.Time{} + return nil + } + d.Time = parsed + d.Valid = true + return nil +} diff --git a/database/csvx/delimited.go b/database/csvx/delimited.go new file mode 100644 index 0000000..2ec5d4a --- /dev/null +++ b/database/csvx/delimited.go @@ -0,0 +1,175 @@ +package csvx + +import ( + "database/sql/driver" + "encoding/json" + "fmt" + "strconv" + "strings" +) + +// Delimiter selects the separator used by a delimited value's CSV form. +type Delimiter interface { + Separator() string +} + +// Comma separates values with a comma. +type Comma struct{} + +func (Comma) Separator() string { return "," } + +// Space separates values with whitespace. +type Space struct{} + +func (Space) Separator() string { return " " } + +// Tab separates values with a tab. +type Tab struct{} + +func (Tab) Separator() string { return "\t" } + +// Strings is a string slice with a delimiter-specific CSV form. +// Its SQL and JSON forms are JSON arrays. +type Strings[D Delimiter] []string + +// Value implements driver.Valuer as a JSON array. +func (s Strings[D]) Value() (driver.Value, error) { + if s == nil { + s = Strings[D]{} + } + return json.Marshal([]string(s)) +} + +// Scan implements sql.Scanner from a JSON array. +func (s *Strings[D]) Scan(value any) error { + switch v := value.(type) { + case nil: + *s = nil + return nil + case []byte: + return s.scanJSON(v) + case string: + return s.scanJSON([]byte(v)) + case Strings[D]: + *s = append((*s)[:0], v...) + return nil + default: + return fmt.Errorf("csvx.Strings: cannot scan %T", value) + } +} + +func (s *Strings[D]) scanJSON(data []byte) error { + if len(data) == 0 { + *s = nil + return nil + } + var values []string + if err := json.Unmarshal(data, &values); err != nil { + return fmt.Errorf("csvx.Strings: %w", err) + } + *s = values + return nil +} + +// MarshalCSV implements csvutil.Marshaler. +func (s Strings[D]) MarshalCSV() ([]byte, error) { + if len(s) == 0 { + return nil, nil + } + var delimiter D + return []byte(strings.Join(s, delimiter.Separator())), nil +} + +// UnmarshalCSV implements csvutil.Unmarshaler. +func (s *Strings[D]) UnmarshalCSV(data []byte) error { + if len(data) == 0 { + *s = nil + return nil + } + var delimiter D + if delimiter.Separator() == (Space{}).Separator() { + *s = Strings[D](strings.Fields(string(data))) + return nil + } + *s = Strings[D](strings.Split(string(data), delimiter.Separator())) + return nil +} + +// Ints is an int64 slice with a delimiter-specific CSV form. +// Its SQL and JSON forms are JSON arrays. +type Ints[D Delimiter] []int64 + +// Value implements driver.Valuer as a JSON array. +func (s Ints[D]) Value() (driver.Value, error) { + if s == nil { + s = Ints[D]{} + } + return json.Marshal([]int64(s)) +} + +// Scan implements sql.Scanner from a JSON array. +func (s *Ints[D]) Scan(value any) error { + switch v := value.(type) { + case nil: + *s = nil + return nil + case []byte: + return s.scanJSON(v) + case string: + return s.scanJSON([]byte(v)) + case Ints[D]: + *s = append((*s)[:0], v...) + return nil + default: + return fmt.Errorf("csvx.Ints: cannot scan %T", value) + } +} + +func (s *Ints[D]) scanJSON(data []byte) error { + if len(data) == 0 { + *s = nil + return nil + } + var values []int64 + if err := json.Unmarshal(data, &values); err != nil { + return fmt.Errorf("csvx.Ints: %w", err) + } + *s = values + return nil +} + +// MarshalCSV implements csvutil.Marshaler. +func (s Ints[D]) MarshalCSV() ([]byte, error) { + if len(s) == 0 { + return nil, nil + } + var delimiter D + values := make([]string, len(s)) + for i, value := range s { + values[i] = strconv.FormatInt(value, 10) + } + return []byte(strings.Join(values, delimiter.Separator())), nil +} + +// UnmarshalCSV implements csvutil.Unmarshaler. +func (s *Ints[D]) UnmarshalCSV(data []byte) error { + if len(data) == 0 { + *s = nil + return nil + } + var delimiter D + parts := strings.Split(string(data), delimiter.Separator()) + if delimiter.Separator() == (Space{}).Separator() { + parts = strings.Fields(string(data)) + } + values := make(Ints[D], len(parts)) + for i, part := range parts { + value, err := strconv.ParseInt(strings.TrimSpace(part), 10, 64) + if err != nil { + return fmt.Errorf("csvx.Ints: parse %q: %w", part, err) + } + values[i] = value + } + *s = values + return nil +} diff --git a/database/csvx/delimited_test.go b/database/csvx/delimited_test.go new file mode 100644 index 0000000..750f1b6 --- /dev/null +++ b/database/csvx/delimited_test.go @@ -0,0 +1,68 @@ +package csvx + +import ( + "bytes" + "encoding/csv" + "testing" +) + +func TestStringsCSV(t *testing.T) { + var values Strings[Comma] + if err := values.UnmarshalCSV([]byte("admin,hr,pm")); err != nil { + t.Fatal(err) + } + data, err := values.MarshalCSV() + if err != nil { + t.Fatal(err) + } + if got := string(data); got != "admin,hr,pm" { + t.Fatalf("CSV = %q", got) + } + + var spaced Strings[Space] + if err := spaced.UnmarshalCSV([]byte("admin hr\tpm")); err != nil { + t.Fatal(err) + } + if len(spaced) != 3 || spaced[1] != "hr" { + t.Fatalf("space values = %#v", spaced) + } +} + +func TestIntsJSONAndSQL(t *testing.T) { + values := Ints[Comma]{1, 2, 3} + got, err := values.Value() + if err != nil { + t.Fatal(err) + } + if string(got.([]byte)) != `[1,2,3]` { + t.Fatalf("Value = %q", got) + } + + var scanned Ints[Comma] + if err := scanned.Scan([]byte(`[1,2,3]`)); err != nil { + t.Fatal(err) + } + if len(scanned) != 3 || scanned[2] != 3 { + t.Fatalf("scanned = %#v", scanned) + } + if err := scanned.UnmarshalCSV([]byte("4,5,6")); err != nil { + t.Fatal(err) + } + if len(scanned) != 3 || scanned[0] != 4 { + t.Fatalf("CSV values = %#v", scanned) + } +} + +func TestDelimitedCSVUsesOuterQuoting(t *testing.T) { + var data bytes.Buffer + writer := csv.NewWriter(&data) + writer.Comma = '\t' + if err := SerializeCSV(writer, []struct { + Values Strings[Tab] `json:"values"` + }{{Values: Strings[Tab]{"a", "b"}}}); err != nil { + t.Fatal(err) + } + if got := data.String(); got != "values\n\"a\tb\"\n" { + t.Fatalf("TSV = %q", got) + } +} diff --git a/database/csvx/doc.go b/database/csvx/doc.go new file mode 100644 index 0000000..bd796cd --- /dev/null +++ b/database/csvx/doc.go @@ -0,0 +1,9 @@ +// Package csvx provides nullable row-oriented data types that bridge +// CSV, JSON, and SQL. Each type implements MarshalCSV/UnmarshalCSV, +// MarshalJSON/UnmarshalJSON, and driver.Value/Scan so the same struct +// can be read from a sheet, written to a file, and scanned from a database. +// +// It also provides generic Parse/Serialize functions for CSV, TSV, JSON, +// and JSONL, delimiter-specific string and integer slices, plus text +// normalization helpers (NormalizeEmail, NormalizePhone). +package csvx diff --git a/database/csvx/encoding.go b/database/csvx/encoding.go new file mode 100644 index 0000000..b7743c7 --- /dev/null +++ b/database/csvx/encoding.go @@ -0,0 +1,172 @@ +package csvx + +import ( + "encoding/csv" + "encoding/json" + "fmt" + "io" + + "github.com/jszwec/csvutil" +) + +// Parse parses records from an io.Reader in the requested format. +func Parse[T any](r io.Reader, format string) ([]T, error) { + switch format { + case "tab", "tsv": + cr := csv.NewReader(r) + cr.Comma = '\t' + cr.FieldsPerRecord = -1 + return ParseCSV[T](cr) + case "csv": + cr := csv.NewReader(r) + cr.Comma = ',' + cr.FieldsPerRecord = -1 + return ParseCSV[T](cr) + case "json", "pretty": + return ParseJSON[T](r) + case "jsonl", "ndjson": + return ParseJSONL[T](r) + default: + return nil, fmt.Errorf("unknown format: %s", format) + } +} + +// ParseCSV reads CSV/TSV from a csvutil Reader using csvutil streaming. +// The reader must already have its delimiter configured. +// Uses json struct tags and csvx marshalers for type handling. +func ParseCSV[T any](r csvutil.Reader) ([]T, error) { + dec, err := csvutil.NewDecoder(r) + if err != nil { + return nil, err + } + dec.Tag = "json" + dec.WithUnmarshalers(Unmarshalers()) + + var items []T + for { + var item T + if err := dec.Decode(&item); err != nil { + if err == io.EOF { + break + } + return nil, err + } + items = append(items, item) + } + return items, nil +} + +// ParseJSONWithEnvelope decodes a JSON envelope from r into T. +// T should be a struct with json tags matching the envelope fields +// (e.g. {"groups": [...], "kind": "...", "etag": "..."}). +// Returns zero value, nil on empty input (io.EOF on first decode). +func ParseJSONWithEnvelope[T any](r io.Reader) (T, error) { + var resp T + dec := json.NewDecoder(r) + if err := dec.Decode(&resp); err != nil { + if err == io.EOF { + var zero T + return zero, nil + } + var zero T + return zero, fmt.Errorf("parse JSON envelope: %w", err) + } + return resp, nil +} + +// ParseJSON decodes a JSON array [{…}, {…}] from r into a slice of T. +// Returns nil, nil on empty input (io.EOF on first decode). +func ParseJSON[T any](r io.Reader) ([]T, error) { + var items []T + dec := json.NewDecoder(r) + if err := dec.Decode(&items); err != nil { + if err == io.EOF { + return nil, nil + } + return nil, err + } + return items, nil +} + +// ParseJSONL reads JSONL (one JSON object per line) from r into a slice of T. +// Each line is decoded independently; decoding stops at io.EOF. +func ParseJSONL[T any](r io.Reader) ([]T, error) { + var items []T + dec := json.NewDecoder(r) + for { + var item T + if err := dec.Decode(&item); err != nil { + if err == io.EOF { + break + } + return nil, fmt.Errorf("decode JSONL: %w", err) + } + items = append(items, item) + } + return items, nil +} + +// Serialize writes values in the specified format. +func Serialize[T any](w io.Writer, values []T, format string) error { + var comma = ',' + switch format { + case "tab", "tsv": + comma = '\t' + fallthrough + case "csv": + cw := csv.NewWriter(w) + cw.Comma = comma + return SerializeCSV(cw, values) + case "jsonl", "ndjson": + enc := json.NewEncoder(w) + return SerializeJSONL(enc, values) + case "pretty": + fallthrough + case "json": + enc := json.NewEncoder(w) + if format == "pretty" { + enc.SetIndent("", " ") + } + if err := SerializeJSON(enc, values); err != nil { + return err + } + _, err := w.Write([]byte("\n")) + return err + default: + return fmt.Errorf("unknown format: %s", format) + } +} + +// SerializeCSV writes items as CSV/TSV to a csv.Writer using json struct tags +// and csvx marshalers. The caller must have configured Comma (or left it at +// the default comma). +func SerializeCSV[T any](cw *csv.Writer, items []T) error { + enc := csvutil.NewEncoder(cw) + enc.Tag = "json" + enc.WithMarshalers(Marshalers()) + if err := enc.Encode(items); err != nil { + return err + } + cw.Flush() + return cw.Error() +} + +// SerializeJSON writes items as a JSON array [{…}, {…}] using the provided +// json.Encoder. +func SerializeJSON[T any](enc *json.Encoder, items []T) error { + if items == nil { + items = []T{} + } + return enc.Encode(items) +} + +// SerializeJSONL writes items as JSONL (one JSON object per line) using the +// provided json.Encoder. +func SerializeJSONL[T any](enc *json.Encoder, items []T) error { + for _, item := range items { + if err := enc.Encode(item); err != nil { + return err + } + } + return nil +} diff --git a/database/csvx/encoding_test.go b/database/csvx/encoding_test.go new file mode 100644 index 0000000..07df4db --- /dev/null +++ b/database/csvx/encoding_test.go @@ -0,0 +1,42 @@ +package csvx + +import ( + "bytes" + "encoding/csv" + "encoding/json" + "testing" +) + +type byteRoundTrip struct { + Value []byte `json:"value"` +} + +func TestSerializeJSONNilSliceAsEmptyArray(t *testing.T) { + var data bytes.Buffer + if err := SerializeJSON[struct{}](json.NewEncoder(&data), nil); err != nil { + t.Fatal(err) + } + if got, want := data.String(), "[]\n"; got != want { + t.Fatalf("JSON = %q, want %q", got, want) + } +} + +func TestByteRoundTripPrefersHex(t *testing.T) { + want := []byte{0x79, 0xf7, 0x5e, 0xdd, 0xce, 0xfe, 0x5d, 0xe5} + var data bytes.Buffer + writer := csv.NewWriter(&data) + writer.Comma = '\t' + if err := SerializeCSV(writer, []byteRoundTrip{{Value: want}}); err != nil { + t.Fatal(err) + } + + reader := csv.NewReader(&data) + reader.Comma = '\t' + got, err := ParseCSV[byteRoundTrip](reader) + if err != nil { + t.Fatal(err) + } + if len(got) != 1 || string(got[0].Value) != string(want) { + t.Fatalf("value = %x, want %x", got[0].Value, want) + } +} diff --git a/database/csvx/http.go b/database/csvx/http.go new file mode 100644 index 0000000..fdd3ab5 --- /dev/null +++ b/database/csvx/http.go @@ -0,0 +1,130 @@ +// Copyright 2025 Cogburn Bros +// +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain a copy at https://mozilla.org/MPL/2.0/. + +package csvx + +import ( + "fmt" + "net/http" + "strconv" + "strings" +) + +// Format is a canonical output format. +type Format string + +const ( + // DefaultFormat is csv — project rule: all new routes default to CSV. + DefaultFormat Format = "csv" + FormatCSV Format = "csv" + FormatTSV Format = "tab" + FormatJSON Format = "json" + FormatJSONL Format = "jsonl" +) + +// ContentType maps canonical format to MIME type. +func ContentType(f Format) string { + switch f { + case FormatCSV: + return "text/csv" + case FormatTSV: + return "text/tab-separated-values" + case FormatJSONL: + return "application/x-ndjson" + default: + return "application/json" + } +} + +// normalize maps ?format= values to canonical format names. +func normalize(s string) (Format, bool) { + switch strings.ToLower(strings.TrimSpace(s)) { + case "csv": + return FormatCSV, true + case "tsv", "tab": + return FormatTSV, true + case "json": + return FormatJSON, true + case "ndjson", "jsonl": + return FormatJSONL, true + } + return "", false +} + +// mediaFormat maps Accept media types to canonical format names. +func mediaFormat(media string) (Format, bool) { + switch strings.ToLower(strings.TrimSpace(media)) { + case "text/csv": + return FormatCSV, true + case "text/tab-separated-values": + return FormatTSV, true + case "application/json", "text/json": + return FormatJSON, true + case "application/x-ndjson", "application/jsonl": + return FormatJSONL, true + } + return "", false +} + +// ResolveFormat returns the output format for a request. ?format= has +// priority. Otherwise the Accept header's best q-matched media type is used. +// Per project rule, the default (empty Accept, */*, or no match) is CSV. +// Returns an error only for an explicit unknown ?format= value. +func ResolveFormat(r *http.Request) (Format, error) { + if f := r.URL.Query().Get("format"); f != "" { + format, ok := normalize(f) + if !ok { + return "", fmt.Errorf("unsupported format %q (accepted: csv, tsv, json, jsonl)", f) + } + return format, nil + } + return acceptFormat(r.Header.Get("Accept")) +} + +// acceptFormat inspects the Accept header's quality values and returns the +// best-matching format. Defaults to CSV when the header is empty, wildcarded, +// or has no matching media type. +func acceptFormat(accept string) (Format, error) { + best, bestQ := DefaultFormat, -1.0 + found := false + + for part := range strings.SplitSeq(accept, ",") { + part = strings.TrimSpace(part) + if part == "" { + continue + } + media := part + q := 1.0 + if before, after, ok := strings.Cut(part, ";"); ok { + media = before + for param := range strings.SplitSeq(after, ";") { + param = strings.TrimSpace(param) + if strings.HasPrefix(strings.ToLower(param), "q=") { + if parsed, err := strconv.ParseFloat(strings.TrimSpace(param[2:]), 64); err == nil { + q = parsed + } + } + } + } + media = strings.TrimSpace(media) + + var f Format + var ok bool + if media == "*/*" { + f, ok = DefaultFormat, true + } else { + f, ok = mediaFormat(media) + } + if ok && q > bestQ { + best, bestQ, found = f, q, true + } + } + + if !found { + return DefaultFormat, nil + } + return best, nil +} diff --git a/database/csvx/normalize.go b/database/csvx/normalize.go new file mode 100644 index 0000000..5ff3af6 --- /dev/null +++ b/database/csvx/normalize.go @@ -0,0 +1,44 @@ +// Package csvx provides text normalization and format detection helpers. +package csvx + +import ( + "path/filepath" + "regexp" + "strings" +) + +var phoneStripRe = regexp.MustCompile(`[^\d,#*]`) + +// NormalizePhone strips non-digit characters and formats 10-digit US numbers +// to (xxx) xxx-xxxx. Non-US or non-standard numbers are returned digit-only. +func NormalizePhone(raw string) string { + stripped := phoneStripRe.ReplaceAllString(raw, "") + if len(stripped) == 11 && stripped[0] == '1' { + stripped = stripped[1:] + } + if len(stripped) == 10 { + return "(" + stripped[0:3] + ") " + stripped[3:6] + "-" + stripped[6:10] + } + return stripped +} + +// NormalizeEmail trims whitespace and lowercases. +func NormalizeEmail(raw string) string { + return strings.ToLower(strings.TrimSpace(raw)) +} + +// DetectFormat returns the format from a file extension. +func DetectFormat(filename string) string { + switch ext := strings.ToLower(filepath.Ext(filename)); ext { + case ".tsv", ".tab": + return "tsv" + case ".csv": + return "csv" + case ".json": + return "json" + case ".jsonl", ".ndjson": + return "jsonl" + default: + return "csv" + } +} diff --git a/database/csvx/number.go b/database/csvx/number.go new file mode 100644 index 0000000..2d2bcc9 --- /dev/null +++ b/database/csvx/number.go @@ -0,0 +1,156 @@ +package csvx + +import ( + "database/sql/driver" + "encoding/json" + "fmt" + "strconv" +) + +// Float64 is a nullable float that implements CSV, JSON, and driver interfaces. +type Float64 struct { + Float64 float64 + Valid bool +} + +func (v Float64) Value() (driver.Value, error) { + if !v.Valid { + return nil, nil + } + return v.Float64, nil +} + +func (v *Float64) Scan(value any) error { + if value == nil { + v.Float64, v.Valid = 0, false + return nil + } + switch x := value.(type) { + case float64: + v.Float64, v.Valid = x, true + case int64: + v.Float64, v.Valid = float64(x), true + case []byte: + return v.Scan(string(x)) + case string: + parsed, err := strconv.ParseFloat(x, 64) + if err != nil { + return fmt.Errorf("csvx.Float64: %w", err) + } + v.Float64, v.Valid = parsed, true + case Float64: + *v = x + default: + return fmt.Errorf("csvx.Float64: cannot scan %T", value) + } + return nil +} + +func (v Float64) MarshalCSV() ([]byte, error) { + if !v.Valid { + return nil, nil + } + return []byte(strconv.FormatFloat(v.Float64, 'g', -1, 64)), nil +} + +func (v *Float64) UnmarshalCSV(data []byte) error { + if len(data) == 0 { + v.Float64, v.Valid = 0, false + return nil + } + return v.Scan(string(data)) +} + +func (v Float64) MarshalJSON() ([]byte, error) { + if !v.Valid { + return []byte("null"), nil + } + return json.Marshal(v.Float64) +} + +func (v *Float64) UnmarshalJSON(data []byte) error { + if string(data) == "null" { + v.Float64, v.Valid = 0, false + return nil + } + var parsed float64 + if err := json.Unmarshal(data, &parsed); err != nil { + return err + } + v.Float64, v.Valid = parsed, true + return nil +} + +// Int64 is a nullable integer that implements CSV, JSON, and driver interfaces. +type Int64 struct { + Int64 int64 + Valid bool +} + +func (v Int64) Value() (driver.Value, error) { + if !v.Valid { + return nil, nil + } + return v.Int64, nil +} + +func (v *Int64) Scan(value any) error { + if value == nil { + v.Int64, v.Valid = 0, false + return nil + } + switch x := value.(type) { + case int64: + v.Int64, v.Valid = x, true + case int: + v.Int64, v.Valid = int64(x), true + case []byte: + return v.Scan(string(x)) + case string: + parsed, err := strconv.ParseInt(x, 10, 64) + if err != nil { + return fmt.Errorf("csvx.Int64: %w", err) + } + v.Int64, v.Valid = parsed, true + case Int64: + *v = x + default: + return fmt.Errorf("csvx.Int64: cannot scan %T", value) + } + return nil +} + +func (v Int64) MarshalCSV() ([]byte, error) { + if !v.Valid { + return nil, nil + } + return []byte(strconv.FormatInt(v.Int64, 10)), nil +} + +func (v *Int64) UnmarshalCSV(data []byte) error { + if len(data) == 0 { + v.Int64, v.Valid = 0, false + return nil + } + return v.Scan(string(data)) +} + +func (v Int64) MarshalJSON() ([]byte, error) { + if !v.Valid { + return []byte("null"), nil + } + return json.Marshal(v.Int64) +} + +func (v *Int64) UnmarshalJSON(data []byte) error { + if string(data) == "null" { + v.Int64, v.Valid = 0, false + return nil + } + var parsed int64 + if err := json.Unmarshal(data, &parsed); err != nil { + return err + } + v.Int64, v.Valid = parsed, true + return nil +} diff --git a/database/csvx/rawjson.go b/database/csvx/rawjson.go new file mode 100644 index 0000000..cbf18a1 --- /dev/null +++ b/database/csvx/rawjson.go @@ -0,0 +1,42 @@ +package csvx + +import ( + "bytes" + "encoding/json" + "fmt" +) + +// RawJSON holds a JSON document as raw bytes. CSV/TSV encode it as JSON +// text, JSON encoding passes it through unchanged, and jsonb binds it as +// bytes. +type RawJSON json.RawMessage + +// MarshalJSON renders the stored document verbatim. +func (r RawJSON) MarshalJSON() ([]byte, error) { + if len(bytes.TrimSpace(r)) == 0 { + return []byte("null"), nil + } + return r, nil +} + +// UnmarshalJSON stores the document verbatim after validation. +func (r *RawJSON) UnmarshalJSON(data []byte) error { + if len(bytes.TrimSpace(data)) == 0 { + *r = nil + return nil + } + if !json.Valid(data) { + return fmt.Errorf("csvx: not valid JSON: %q", bytes.TrimSpace(data)) + } + *r = RawJSON(append([]byte(nil), data...)) + return nil +} + +// MarshalCSV renders the document as JSON text. +func (r RawJSON) MarshalCSV() ([]byte, error) { return r, nil } + +// UnmarshalCSV stores the field as the JSON document. +func (r *RawJSON) UnmarshalCSV(data []byte) error { return r.UnmarshalJSON(data) } + +// String returns the document as a string. +func (r RawJSON) String() string { return string(r) } diff --git a/database/csvx/text.go b/database/csvx/text.go new file mode 100644 index 0000000..7c4b79b --- /dev/null +++ b/database/csvx/text.go @@ -0,0 +1,89 @@ +package csvx + +import ( + "database/sql/driver" + "encoding/json" + "fmt" +) + +// Text is a nullable string that implements CSV, JSON, and driver interfaces. +type Text struct { + String string + Valid bool +} + +func (t Text) IsNil() bool { return !t.Valid } + +func (t Text) Value() (driver.Value, error) { + if !t.Valid { + return nil, nil + } + return t.String, nil +} + +func (t *Text) Scan(value any) error { + if value == nil { + t.Valid = false + t.String = "" + return nil + } + switch v := value.(type) { + case string: + t.String = v + t.Valid = true + case []byte: + t.String = string(v) + t.Valid = true + case Text: + *t = v + default: + return fmt.Errorf("csvx.Text: cannot scan %T", value) + } + return nil +} + +func (t Text) MarshalCSV() ([]byte, error) { + if !t.Valid { + return nil, nil + } + return []byte(t.String), nil +} + +func (t *Text) UnmarshalCSV(data []byte) error { + s := string(data) + if s == "" { + t.Valid = false + t.String = "" + return nil + } + t.String = s + t.Valid = true + return nil +} + +func (t Text) MarshalJSON() ([]byte, error) { + if !t.Valid { + return []byte("null"), nil + } + return json.Marshal(t.String) +} + +func (t *Text) UnmarshalJSON(data []byte) error { + if string(data) == "null" { + t.Valid = false + t.String = "" + return nil + } + var s string + if err := json.Unmarshal(data, &s); err != nil { + return err + } + if s == "" { + t.Valid = false + t.String = "" + return nil + } + t.String = s + t.Valid = true + return nil +} diff --git a/database/csvx/timestamp.go b/database/csvx/timestamp.go new file mode 100644 index 0000000..97b479f --- /dev/null +++ b/database/csvx/timestamp.go @@ -0,0 +1,120 @@ +package csvx + +import ( + "database/sql/driver" + "encoding/json" + "fmt" + "time" +) + +// Timestamp is a nullable timestamp (without timezone) that implements CSV, +// JSON, and driver interfaces. +type Timestamp struct { + Time time.Time + Valid bool +} + +func (t Timestamp) IsNil() bool { return !t.Valid } + +func (t Timestamp) Value() (driver.Value, error) { + if !t.Valid { + return nil, nil + } + return t.Time, nil +} + +func (t *Timestamp) Scan(value any) error { + if value == nil { + t.Valid = false + t.Time = time.Time{} + return nil + } + switch v := value.(type) { + case time.Time: + t.Time = v + t.Valid = true + case string: + parsed, err := parseTimestamp(v) + if err != nil { + t.Valid = false + t.Time = time.Time{} + return nil + } + t.Time = parsed + t.Valid = true + case []byte: + return t.Scan(string(v)) + case Timestamp: + *t = v + default: + return fmt.Errorf("csvx.Timestamp: cannot scan %T", value) + } + return nil +} + +func (t Timestamp) MarshalCSV() ([]byte, error) { + if !t.Valid { + return nil, nil + } + return []byte(t.Time.Format(time.RFC3339Nano)), nil +} + +func (t *Timestamp) UnmarshalCSV(data []byte) error { + s := string(data) + if s == "" { + t.Valid = false + t.Time = time.Time{} + return nil + } + parsed, err := parseTimestamp(s) + if err != nil { + t.Valid = false + t.Time = time.Time{} + return nil + } + t.Time = parsed + t.Valid = true + return nil +} + +func (t Timestamp) MarshalJSON() ([]byte, error) { + if !t.Valid { + return []byte("null"), nil + } + return json.Marshal(t.Time.Format(time.RFC3339Nano)) +} + +func (t *Timestamp) UnmarshalJSON(data []byte) error { + if string(data) == "null" { + t.Valid = false + t.Time = time.Time{} + return nil + } + var s string + if err := json.Unmarshal(data, &s); err != nil { + return err + } + if s == "" { + t.Valid = false + t.Time = time.Time{} + return nil + } + parsed, err := parseTimestamp(s) + if err != nil { + t.Valid = false + t.Time = time.Time{} + return nil + } + t.Time = parsed + t.Valid = true + return nil +} + +func parseTimestamp(s string) (time.Time, error) { + for _, layout := range []string{time.RFC3339Nano, time.RFC3339, "2006-01-02 15:04:05", "2006-01-02"} { + if parsed, err := time.Parse(layout, s); err == nil { + return parsed, nil + } + } + return time.Time{}, fmt.Errorf("csvx.Timestamp: unrecognized time %q", s) +} diff --git a/database/csvx/timestamptz.go b/database/csvx/timestamptz.go new file mode 100644 index 0000000..c5092c7 --- /dev/null +++ b/database/csvx/timestamptz.go @@ -0,0 +1,111 @@ +package csvx + +import ( + "database/sql/driver" + "encoding/json" + "fmt" + "time" +) + +// Timestamptz is a nullable timestamp with timezone that implements CSV, +// JSON, and driver interfaces. Time is stored in UTC. +type Timestamptz struct { + Time time.Time + Valid bool +} + +func (t Timestamptz) IsNil() bool { return !t.Valid } + +func (t Timestamptz) Value() (driver.Value, error) { + if !t.Valid { + return nil, nil + } + return t.Time.UTC(), nil +} + +func (t *Timestamptz) Scan(value any) error { + if value == nil { + t.Valid = false + t.Time = time.Time{} + return nil + } + switch v := value.(type) { + case time.Time: + t.Time = v.UTC() + t.Valid = true + case string: + parsed, err := parseTimestamp(v) + if err != nil { + t.Valid = false + t.Time = time.Time{} + return nil + } + t.Time = parsed.UTC() + t.Valid = true + case []byte: + return t.Scan(string(v)) + case Timestamptz: + *t = v + default: + return fmt.Errorf("csvx.Timestamptz: cannot scan %T", value) + } + return nil +} + +func (t Timestamptz) MarshalCSV() ([]byte, error) { + if !t.Valid { + return nil, nil + } + return []byte(t.Time.UTC().Format(time.RFC3339Nano)), nil +} + +func (t *Timestamptz) UnmarshalCSV(data []byte) error { + s := string(data) + if s == "" { + t.Valid = false + t.Time = time.Time{} + return nil + } + parsed, err := parseTimestamp(s) + if err != nil { + t.Valid = false + t.Time = time.Time{} + return nil + } + t.Time = parsed.UTC() + t.Valid = true + return nil +} + +func (t Timestamptz) MarshalJSON() ([]byte, error) { + if !t.Valid { + return []byte("null"), nil + } + return json.Marshal(t.Time.UTC().Format(time.RFC3339Nano)) +} + +func (t *Timestamptz) UnmarshalJSON(data []byte) error { + if string(data) == "null" { + t.Valid = false + t.Time = time.Time{} + return nil + } + var s string + if err := json.Unmarshal(data, &s); err != nil { + return err + } + if s == "" { + t.Valid = false + t.Time = time.Time{} + return nil + } + parsed, err := parseTimestamp(s) + if err != nil { + t.Valid = false + t.Time = time.Time{} + return nil + } + t.Time = parsed.UTC() + t.Valid = true + return nil +} diff --git a/database/csvx/uuid.go b/database/csvx/uuid.go new file mode 100644 index 0000000..2134ec5 --- /dev/null +++ b/database/csvx/uuid.go @@ -0,0 +1,127 @@ +package csvx + +import ( + "database/sql/driver" + "encoding/json" + "fmt" + + "github.com/google/uuid" +) + +// UUID is a nullable UUID that implements CSV, JSON, and driver interfaces. +type UUID struct { + Bytes [16]byte + Valid bool +} + +func (u UUID) String() string { + if !u.Valid { + return "" + } + return uuid.UUID(u.Bytes).String() +} + +func (u UUID) IsNil() bool { return !u.Valid } + +func (u UUID) Value() (driver.Value, error) { + if !u.Valid { + return nil, nil + } + return u.String(), nil +} + +func (u *UUID) Scan(value any) error { + if value == nil { + u.Valid = false + u.Bytes = [16]byte{} + return nil + } + switch v := value.(type) { + case string: + parsed, err := uuid.Parse(v) + if err != nil { + u.Valid = false + u.Bytes = [16]byte{} + return nil + } + copy(u.Bytes[:], parsed[:]) + u.Valid = true + case []byte: + if len(v) == 16 { + copy(u.Bytes[:], v) + u.Valid = true + return nil + } + parsed, err := uuid.Parse(string(v)) + if err != nil { + u.Valid = false + u.Bytes = [16]byte{} + return nil + } + copy(u.Bytes[:], parsed[:]) + u.Valid = true + case UUID: + *u = v + default: + return fmt.Errorf("csvx.UUID: cannot scan %T", value) + } + return nil +} + +func (u UUID) MarshalCSV() ([]byte, error) { + if !u.Valid { + return nil, nil + } + return []byte(u.String()), nil +} + +func (u *UUID) UnmarshalCSV(data []byte) error { + s := string(data) + if s == "" { + u.Valid = false + u.Bytes = [16]byte{} + return nil + } + parsed, err := uuid.Parse(s) + if err != nil { + u.Valid = false + u.Bytes = [16]byte{} + return nil + } + copy(u.Bytes[:], parsed[:]) + u.Valid = true + return nil +} + +func (u UUID) MarshalJSON() ([]byte, error) { + if !u.Valid { + return []byte("null"), nil + } + return json.Marshal(u.String()) +} + +func (u *UUID) UnmarshalJSON(data []byte) error { + if string(data) == "null" { + u.Valid = false + u.Bytes = [16]byte{} + return nil + } + var s string + if err := json.Unmarshal(data, &s); err != nil { + return err + } + if s == "" { + u.Valid = false + u.Bytes = [16]byte{} + return nil + } + parsed, err := uuid.Parse(s) + if err != nil { + u.Valid = false + u.Bytes = [16]byte{} + return nil + } + copy(u.Bytes[:], parsed[:]) + u.Valid = true + return nil +} diff --git a/database/go.mod b/database/go.mod new file mode 100644 index 0000000..0236f4f --- /dev/null +++ b/database/go.mod @@ -0,0 +1,19 @@ +module github.com/therootcompany/golib/database + +go 1.26.5 + +require ( + github.com/google/uuid v1.6.0 + github.com/jackc/pgx/v5 v5.10.0 + github.com/jszwec/csvutil v1.10.0 + github.com/therootcompany/golib/database/sqlmigrate v1.0.3 + github.com/therootcompany/golib/database/sqlmigrate/pgmigrate v1.0.5 + github.com/therootcompany/golib/io/transform/gsheet2csv v1.1.0 +) + +require ( + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/therootcompany/golib/https v0.9.0 // indirect + golang.org/x/text v0.29.0 // indirect +) diff --git a/database/go.sum b/database/go.sum new file mode 100644 index 0000000..b3da5f9 --- /dev/null +++ b/database/go.sum @@ -0,0 +1,38 @@ +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/jszwec/csvutil v1.10.0 h1:upMDUxhQKqZ5ZDCs/wy+8Kib8rZR8I8lOR34yJkdqhI= +github.com/jszwec/csvutil v1.10.0/go.mod h1:/E4ONrmGkwmWsk9ae9jpXnv9QT8pLHEPcCirMFhxG9I= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/therootcompany/golib/database/sqlmigrate v1.0.3 h1:/6cun+EZoJ1eSNM8pQw8ysJajCGVdMOISihSI5Ko0no= +github.com/therootcompany/golib/database/sqlmigrate v1.0.3/go.mod h1:7PQUjwT78Hx+SftcIKI2PH4zSFlrSO0V9h618PJqC38= +github.com/therootcompany/golib/database/sqlmigrate/pgmigrate v1.0.5 h1:v28gNvbO/eVwAntfS+LEaTC/eHs+J+rUix3Xl9Ty9qo= +github.com/therootcompany/golib/database/sqlmigrate/pgmigrate v1.0.5/go.mod h1:uIxJrM0dM+bzKIOWDgf6rWGYE0B+7cPA3bAbuNheaWA= +github.com/therootcompany/golib/https v0.9.0 h1:D9gh0K6ljkvF805JUbylITn8SHPMo08q5PglQGXsK8c= +github.com/therootcompany/golib/https v0.9.0/go.mod h1:ilExqR/Uh7oB36oy9mwuNhANcFh+zdCfD3uJiZ11YAY= +github.com/therootcompany/golib/io/transform/gsheet2csv v1.1.0 h1:R5zAPun5p3JPlNfcMD9es4LD9sqh6nWDNTflZBeeUJo= +github.com/therootcompany/golib/io/transform/gsheet2csv v1.1.0/go.mod h1:CsntOAhhIMdzSAYuwuRZtfkdoiER9+TfSgYOIDWN71o= +golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= +golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk= +golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/database/pgdb/conn.go b/database/pgdb/conn.go new file mode 100644 index 0000000..c8a6423 --- /dev/null +++ b/database/pgdb/conn.go @@ -0,0 +1,172 @@ +// Package pgdb provides a generic PostgreSQL database adapter. +// It can serve as both a source (Fetcher) and destination (Updater), +// enabling pipelines like gsheet -> pg or pg -> tsv. +// +// Store.DB accepts either *pgx.Conn (CLI, tests) or *pgxpool.Pool (servers). +// Import "github.com/jackc/pgx/v5/pgxpool" if using a pool. +package pgdb + +import ( + "context" + "fmt" + "io/fs" + "time" + + "github.com/therootcompany/golib/database/rowsync" + + "github.com/jackc/pgx/v5" + "github.com/therootcompany/golib/database/sqlmigrate" + "github.com/therootcompany/golib/database/sqlmigrate/pgmigrate" +) + +// DBT is the minimum interface satisfied by both *pgx.Conn and *pgxpool.Pool. +type DBT interface { + Ping(context.Context) error + Begin(context.Context) (pgx.Tx, error) +} + +// Connect establishes a single PostgreSQL connection with a 5-second timeout. +// For servers, prefer pgxpool.New and pass the pool as DBT. +func Connect(parent context.Context, pgURL string) (*pgx.Conn, error) { + ctx, cancel := context.WithTimeout(parent, 5*time.Second) + defer cancel() + conn, err := pgx.Connect(ctx, pgURL) + if err != nil { + return nil, fmt.Errorf("connect PostgreSQL: %w", err) + } + return conn, nil +} + +// Migrate runs pending migrations from an fs.FS on the connection. +// Returns messages about what was applied, or nil if nothing changed. +func Migrate(ctx context.Context, conn *pgx.Conn, migrationsFS fs.FS) ([]rowsync.Message, error) { + ddls, err := sqlmigrate.Collect(migrationsFS, ".") + if err != nil { + return nil, fmt.Errorf("collect migrations: %w", err) + } + runner := pgmigrate.New(conn) + applied, err := sqlmigrate.Latest(ctx, runner, ddls) + if err != nil { + return nil, fmt.Errorf("apply migrations: %w", err) + } + if len(applied) == 0 { + return nil, nil + } + msgs := make([]rowsync.Message, 0, len(applied)+1) + for _, m := range applied { + msgs = append(msgs, rowsync.Message{String: m.Name, Type: "debug"}) + } + msgs = append(msgs, rowsync.Message{ + String: fmt.Sprintf("applied %d migration(s)", len(applied)), + Type: "info", + }) + return msgs, nil +} + +// Store is a generic PostgreSQL Fetcher and Updater. +// The application provides FetchFunc, UpsertFunc, and optionally DeleteFunc +// to define the SQL. +// Capture the connection or pool in the function closures: +// +// store := &dbpg.Store[MyRow]{ +// DB: pool, +// FetchFunc: func(ctx context.Context) ([]MyRow, error) { +// return queries.MyRowList(ctx) +// }, +// UpsertFunc: func(ctx context.Context, tx pgx.Tx, row MyRow) error { +// return queries.New(tx).MyRowUpsert(ctx, row) +// }, +// DeleteFunc: func(ctx context.Context, tx pgx.Tx, row MyRow) error { +// return queries.New(tx).MyRowDelete(ctx, row.ID) +// }, +// } +type Store[T any] struct { + DB DBT + Migrations fs.FS // optional; if set, Init runs migrations + FetchFunc func(ctx context.Context) ([]T, error) + UpsertFunc func(ctx context.Context, tx pgx.Tx, row T) error + DeleteFunc func(ctx context.Context, tx pgx.Tx, row T) error // required when a plan carries deletes + Normalize func([]T) +} + +var _ rowsync.Fetcher[struct{}] = (*Store[struct{}])(nil) +var _ rowsync.Updater[struct{}] = (*Store[struct{}])(nil) + +// Init pings the connection and optionally runs migrations. +func (s *Store[T]) Init(ctx context.Context) ([]rowsync.Message, error) { + if s == nil || s.DB == nil { + return nil, fmt.Errorf("PostgreSQL connection is required") + } + if err := s.DB.Ping(ctx); err != nil { + return nil, err + } + if s.Migrations != nil { + // Migrations require a *pgx.Conn (pgmigrate uses conn-specific APIs). + conn, ok := s.DB.(*pgx.Conn) + if !ok { + return nil, fmt.Errorf("migrations require a *pgx.Conn, got %T", s.DB) + } + msgs, err := Migrate(ctx, conn, s.Migrations) + if err != nil { + return nil, err + } + for i := range msgs { + msgs[i].Type = "info" + } + return msgs, nil + } + return []rowsync.Message{{String: "PostgreSQL", Type: "debug"}}, nil +} + +// Fetch reads all rows from PostgreSQL using FetchFunc. +func (s *Store[T]) Fetch(ctx context.Context) ([]T, error) { + if s == nil || s.DB == nil { + return nil, fmt.Errorf("PostgreSQL connection is required") + } + if s.FetchFunc == nil { + return nil, fmt.Errorf("FetchFunc is required") + } + rows, err := s.FetchFunc(ctx) + if err != nil { + return nil, err + } + if s.Normalize != nil { + s.Normalize(rows) + } + return rows, nil +} + +// Update applies a plan within one transaction: insert and update actions +// upsert Action.After, delete actions pass Action.Before to DeleteFunc. +func (s *Store[T]) Update(ctx context.Context, plan rowsync.Plan[T]) error { + if s == nil || s.DB == nil { + return fmt.Errorf("PostgreSQL connection is required") + } + if s.UpsertFunc == nil { + return fmt.Errorf("UpsertFunc is required") + } + if len(plan.Actions) == 0 { + return nil + } + tx, err := s.DB.Begin(ctx) + if err != nil { + return err + } + defer func() { _ = tx.Rollback(ctx) }() + for _, action := range plan.Actions { + switch action.Kind { + case rowsync.InsertAction, rowsync.UpdateAction: + if err := s.UpsertFunc(ctx, tx, *action.After); err != nil { + return err + } + case rowsync.DeleteAction: + if s.DeleteFunc == nil { + return fmt.Errorf("DeleteFunc is required for delete actions") + } + if err := s.DeleteFunc(ctx, tx, *action.Before); err != nil { + return err + } + } + } + return tx.Commit(ctx) +} diff --git a/database/rowsync/compose.go b/database/rowsync/compose.go new file mode 100644 index 0000000..9ecaadb --- /dev/null +++ b/database/rowsync/compose.go @@ -0,0 +1,97 @@ +package rowsync + +import ( + "context" + "fmt" +) + +// RunStore executes the fetch → plan → apply pipeline. The planner's Mode is +// the explicit operation choice: UpdateMode is insert-update only, SyncMode +// makes the source authoritative and deletes destination rows the source +// lacks. An empty source is rejected before planning unless the destination +// is also empty, in which case the run is a no-op. +func RunStore[T any]( + ctx context.Context, + input Fetcher[T], + output Updater[T], + planner Planner[T], +) (Result[T], error) { + return runPipeline(ctx, input, output, planner, nil) +} + +// RunMergedStore is RunStore with an explicit merge step for partial +// sources. The planner is forced to UpdateMode: a partial (merged) source is +// never authoritative, so destination rows the source lacks survive. +func RunMergedStore[T any]( + ctx context.Context, + input Fetcher[T], + output Updater[T], + merge Merge[T], + planner Planner[T], +) (Result[T], error) { + if merge == nil { + return Result[T]{}, fmt.Errorf("merge function is required") + } + planner.Mode = UpdateMode + return runPipeline(ctx, input, output, planner, merge) +} + +func runPipeline[T any]( + ctx context.Context, + input Fetcher[T], + output Updater[T], + planner Planner[T], + merge Merge[T], +) (Result[T], error) { + if input == nil { + return Result[T]{}, fmt.Errorf("input fetcher is required") + } + if output == nil { + return Result[T]{}, fmt.Errorf("output updater is required") + } + if planner.Key == nil || planner.Equal == nil { + return Result[T]{}, fmt.Errorf("planner key and equal functions are required") + } + if _, err := input.Init(ctx); err != nil { + return Result[T]{}, fmt.Errorf("init input: %w", err) + } + if _, err := output.Init(ctx); err != nil { + return Result[T]{}, fmt.Errorf("init output: %w", err) + } + source, err := input.Fetch(ctx) + if err != nil { + return Result[T]{}, fmt.Errorf("fetch input: %w", err) + } + previous, err := output.Fetch(ctx) + if err != nil { + return Result[T]{}, fmt.Errorf("fetch output: %w", err) + } + result := Result[T]{Source: source} + if reporter, ok := input.(InvalidReporter); ok { + result.Invalid = reporter.InvalidCount() + } + if reporter, ok := input.(InvalidRowReporter); ok { + result.InvalidRows = reporter.InvalidRows() + } + current := source + if merge != nil { + current, err = merge(previous, source) + if err != nil { + return Result[T]{}, fmt.Errorf("merge: %w", err) + } + } + if len(current) == 0 && len(previous) == 0 { + // Both sides empty: nothing to plan or apply. + result.Plan = Plan[T]{Mode: planner.Mode} + return result, nil + } + plan, err := planner.Build(previous, current) + if err != nil { + return Result[T]{}, fmt.Errorf("build plan: %w", err) + } + if err := output.Update(ctx, plan); err != nil { + return Result[T]{}, fmt.Errorf("apply plan: %w", err) + } + result.Plan = plan + return result, nil +} diff --git a/database/rowsync/plan.go b/database/rowsync/plan.go new file mode 100644 index 0000000..cb66baf --- /dev/null +++ b/database/rowsync/plan.go @@ -0,0 +1,318 @@ +package rowsync + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "maps" + "sort" +) + +// Mode controls whether a plan may delete destination rows. +type Mode uint8 + +const ( + // UpdateMode inserts and updates rows but never deletes destination rows. + UpdateMode Mode = iota + // SyncMode makes the source authoritative and includes deletes. + SyncMode +) + +// ActionKind identifies one planned destination action. +type ActionKind uint8 + +const ( + InsertAction ActionKind = iota + UpdateAction + DeleteAction +) + +// Action is one deterministic change to apply. +type Action[T any] struct { + Kind ActionKind + Key string + Before *T + After *T + ChangedFields []string +} + +// Plan is the complete result of comparing source and destination rows. +type Plan[T any] struct { + Mode Mode + Previous []T + Current []T + Final []T + Actions []Action[T] +} + +// Counts summarizes planned actions. +type Counts struct { + Inserted int + Updated int + Deleted int +} + +// Count returns the number of actions by kind. +func (p Plan[T]) Count() Counts { + var counts Counts + for _, action := range p.Actions { + switch action.Kind { + case InsertAction: + counts.Inserted++ + case UpdateAction: + counts.Updated++ + case DeleteAction: + counts.Deleted++ + } + } + return counts +} + +// ActionSummary is safe for audit output. It never contains the source key. +type ActionSummary struct { + Kind ActionKind + KeyDigest string + ChangedFields []string +} + +// AuditRecord maps one plan action to safe audit metadata: action kind, key +// digest, changed fields, and digests of the complete old and new rows. Raw +// row data (which may hold secrets, tokens, or authorization codes) is never +// exposed; consumers needing full rows hold the plan itself. +type AuditRecord struct { + Kind ActionKind + KeyDigest string + ChangedFields []string + BeforeDigest string // sha256 of the complete previous row; "" for inserts + AfterDigest string // sha256 of the complete new row; "" for deletes +} + +// AuditRecords returns one safe audit record per planned action, in plan +// order. Row digests are stable for value-type rows within a build. +func (p Plan[T]) AuditRecords() []AuditRecord { + result := make([]AuditRecord, 0, len(p.Actions)) + for _, action := range p.Actions { + digest := sha256.Sum256([]byte(action.Key)) + record := AuditRecord{ + Kind: action.Kind, + KeyDigest: hex.EncodeToString(digest[:6]), + ChangedFields: append([]string(nil), action.ChangedFields...), + } + if action.Before != nil { + record.BeforeDigest = rowDigest(*action.Before) + } + if action.After != nil { + record.AfterDigest = rowDigest(*action.After) + } + result = append(result, record) + } + return result +} + +func rowDigest[T any](row T) string { + digest := sha256.Sum256(fmt.Append(nil, row)) + return hex.EncodeToString(digest[:]) +} + +// RedactedActions returns deterministic action metadata without exposing keys. +func (p Plan[T]) RedactedActions() []ActionSummary { + result := make([]ActionSummary, 0, len(p.Actions)) + for _, action := range p.Actions { + digest := sha256.Sum256([]byte(action.Key)) + result = append(result, ActionSummary{ + Kind: action.Kind, + KeyDigest: hex.EncodeToString(digest[:6]), + ChangedFields: append([]string(nil), action.ChangedFields...), + }) + } + return result +} + +// Planner compares rows by a stable key. +type Planner[T any] struct { + Mode Mode + Key func(T) string + Equal func(T, T) bool + ChangedFields func(T, T) []string +} + +// Build creates a deterministic action plan. Empty or duplicate keys fail. +func (p Planner[T]) Build(previous, current []T) (Plan[T], error) { + if len(current) == 0 { + return Plan[T]{}, fmt.Errorf("source returned zero rows") + } + if p.Key == nil { + return Plan[T]{}, fmt.Errorf("key function is required") + } + if p.Equal == nil { + return Plan[T]{}, fmt.Errorf("equal function is required") + } + before, err := indexRows(previous, p.Key) + if err != nil { + return Plan[T]{}, fmt.Errorf("index previous rows: %w", err) + } + after, err := indexRows(current, p.Key) + if err != nil { + return Plan[T]{}, fmt.Errorf("index current rows: %w", err) + } + keys := make([]string, 0, len(before)+len(after)) + seen := make(map[string]struct{}, len(before)+len(after)) + for key := range before { + seen[key] = struct{}{} + keys = append(keys, key) + } + for key := range after { + if _, ok := seen[key]; !ok { + keys = append(keys, key) + } + } + sort.Strings(keys) + + plan := Plan[T]{Mode: p.Mode, Previous: append([]T(nil), previous...), Current: append([]T(nil), current...)} + final := make(map[string]T, len(before)+len(after)) + maps.Copy(final, before) + maps.Copy(final, after) + if p.Mode == SyncMode { + for key := range before { + if _, ok := after[key]; !ok { + delete(final, key) + } + } + } + for _, key := range keys { + oldRow, hadOld := before[key] + newRow, hasNew := after[key] + switch { + case !hadOld && hasNew: + row := newRow + plan.Actions = append(plan.Actions, Action[T]{Kind: InsertAction, Key: key, After: &row}) + case hadOld && !hasNew && p.Mode == SyncMode: + row := oldRow + plan.Actions = append(plan.Actions, Action[T]{Kind: DeleteAction, Key: key, Before: &row}) + case hadOld && hasNew && !p.Equal(oldRow, newRow): + oldCopy, newCopy := oldRow, newRow + action := Action[T]{Kind: UpdateAction, Key: key, Before: &oldCopy, After: &newCopy} + if p.ChangedFields != nil { + action.ChangedFields = append([]string(nil), p.ChangedFields(oldRow, newRow)...) + } + plan.Actions = append(plan.Actions, action) + } + } + sort.Slice(plan.Actions, func(i, j int) bool { + left, right := plan.Actions[i], plan.Actions[j] + if actionOrder(left.Kind) != actionOrder(right.Kind) { + return actionOrder(left.Kind) < actionOrder(right.Kind) + } + return left.Key < right.Key + }) + + finalKeys := make([]string, 0, len(final)) + for key := range final { + finalKeys = append(finalKeys, key) + } + sort.Strings(finalKeys) + plan.Final = make([]T, 0, len(finalKeys)) + for _, key := range finalKeys { + plan.Final = append(plan.Final, final[key]) + } + return plan, nil +} + +func actionOrder(kind ActionKind) int { + switch kind { + case DeleteAction: + return 0 + case UpdateAction: + return 1 + case InsertAction: + return 2 + default: + return 3 + } +} + +func indexRows[T any](rows []T, keyFunc func(T) string) (map[string]T, error) { + indexed := make(map[string]T, len(rows)) + for _, row := range rows { + key := keyFunc(row) + if key == "" { + return nil, fmt.Errorf("row has empty key") + } + if _, exists := indexed[key]; exists { + return nil, fmt.Errorf("duplicate key %q", key) + } + indexed[key] = row + } + return indexed, nil +} + +// PlanResult contains source rows and the unapplied plan. +type PlanResult[T any] struct { + Source []T + Plan Plan[T] +} + +// BuildPlan fetches source and destination rows and builds one plan. +// It does not apply the plan. +func BuildPlan[T any](ctx context.Context, input Fetcher[T], output Fetcher[T], planner Planner[T]) (PlanResult[T], error) { + return buildPlan(ctx, input, output, planner, nil) +} + +// BuildMergedPlan is BuildPlan with an explicit merge step for partial input. +func BuildMergedPlan[T any]( + ctx context.Context, + input Fetcher[T], + output Fetcher[T], + planner Planner[T], + merge Merge[T], +) (PlanResult[T], error) { + if merge == nil { + return PlanResult[T]{}, fmt.Errorf("merge function is required") + } + return buildPlan(ctx, input, output, planner, merge) +} + +func buildPlan[T any]( + ctx context.Context, + input Fetcher[T], + output Fetcher[T], + planner Planner[T], + merge Merge[T], +) (PlanResult[T], error) { + if input == nil { + return PlanResult[T]{}, fmt.Errorf("input fetcher is required") + } + if output == nil { + return PlanResult[T]{}, fmt.Errorf("output updater is required") + } + if _, err := input.Init(ctx); err != nil { + return PlanResult[T]{}, fmt.Errorf("init input: %w", err) + } + if _, err := output.Init(ctx); err != nil { + return PlanResult[T]{}, fmt.Errorf("init output: %w", err) + } + source, err := input.Fetch(ctx) + if err != nil { + return PlanResult[T]{}, fmt.Errorf("fetch input: %w", err) + } + previous, err := output.Fetch(ctx) + if err != nil { + return PlanResult[T]{}, fmt.Errorf("fetch output: %w", err) + } + current := source + if merge != nil { + current, err = merge(previous, source) + if err != nil { + return PlanResult[T]{}, fmt.Errorf("merge: %w", err) + } + } + if len(current) == 0 && len(previous) == 0 { + return PlanResult[T]{Source: source, Plan: Plan[T]{Mode: planner.Mode}}, nil + } + plan, err := planner.Build(previous, current) + if err != nil { + return PlanResult[T]{}, fmt.Errorf("build plan: %w", err) + } + return PlanResult[T]{Source: source, Plan: plan}, nil +} diff --git a/database/rowsync/plan_test.go b/database/rowsync/plan_test.go new file mode 100644 index 0000000..a113f62 --- /dev/null +++ b/database/rowsync/plan_test.go @@ -0,0 +1,380 @@ +package rowsync + +import ( + "context" + "errors" + "reflect" + "testing" +) + +type planRow struct { + ID string + Value string +} + +func rowPlanner(mode Mode) Planner[planRow] { + return Planner[planRow]{ + Mode: mode, + Key: func(row planRow) string { return row.ID }, + Equal: func(a, b planRow) bool { return a == b }, + ChangedFields: func(a, b planRow) []string { + if a.Value != b.Value { + return []string{"value"} + } + return nil + }, + } +} + +func TestPlannerBuildsDeterministicPlan(t *testing.T) { + plan, err := rowPlanner(UpdateMode).Build([]planRow{{ID: "old", Value: "one"}}, []planRow{{ID: "new", Value: "two"}}) + if err != nil { + t.Fatal(err) + } + if got, want := plan.Count(), (Counts{Inserted: 1}); got != want { + t.Fatalf("counts = %#v, want %#v", got, want) + } + if got := plan.Actions[0].Key; got != "new" { + t.Fatalf("action key = %q, want new", got) + } +} + +func TestPlannerEmitsAllActionKinds(t *testing.T) { + previous := []planRow{ + {ID: "a", Value: "same"}, + {ID: "b", Value: "old"}, + {ID: "c", Value: "gone"}, + } + current := []planRow{ + {ID: "a", Value: "same"}, + {ID: "b", Value: "new"}, + {ID: "d", Value: "added"}, + } + plan, err := rowPlanner(SyncMode).Build(previous, current) + if err != nil { + t.Fatal(err) + } + want := []Action[planRow]{ + {Kind: DeleteAction, Key: "c", Before: &previous[2]}, + {Kind: UpdateAction, Key: "b", Before: &previous[1], After: ¤t[1], ChangedFields: []string{"value"}}, + {Kind: InsertAction, Key: "d", After: ¤t[2]}, + } + if !reflect.DeepEqual(plan.Actions, want) { + t.Fatalf("actions = %#v, want %#v", plan.Actions, want) + } + // Final state: b updated, c removed, d added, a untouched. + wantFinal := []planRow{{ID: "a", Value: "same"}, {ID: "b", Value: "new"}, {ID: "d", Value: "added"}} + if !reflect.DeepEqual(plan.Final, wantFinal) { + t.Fatalf("final = %#v, want %#v", plan.Final, wantFinal) + } +} + +func TestPlannerUpdateModeNeverDeletes(t *testing.T) { + plan, err := rowPlanner(UpdateMode).Build([]planRow{{ID: "keep"}}, []planRow{{ID: "add"}}) + if err != nil { + t.Fatal(err) + } + if got := plan.Count().Deleted; got != 0 { + t.Fatalf("deleted = %d, want 0", got) + } + // Destination-only rows survive in the final state. + wantFinal := []planRow{{ID: "add"}, {ID: "keep"}} + if !reflect.DeepEqual(plan.Final, wantFinal) { + t.Fatalf("final = %#v, want %#v", plan.Final, wantFinal) + } +} + +func TestPlannerSyncModeFinalIsSourceOnly(t *testing.T) { + plan, err := rowPlanner(SyncMode).Build([]planRow{{ID: "keep"}, {ID: "drop"}}, []planRow{{ID: "keep"}, {ID: "add"}}) + if err != nil { + t.Fatal(err) + } + wantFinal := []planRow{{ID: "add"}, {ID: "keep"}} + if !reflect.DeepEqual(plan.Final, wantFinal) { + t.Fatalf("final = %#v, want %#v", plan.Final, wantFinal) + } +} + +func TestPlannerRejectsDuplicateKeys(t *testing.T) { + _, err := rowPlanner(UpdateMode).Build(nil, []planRow{{ID: "x"}, {ID: "x"}}) + if err == nil { + t.Fatal("Build accepted duplicate current keys") + } + _, err = rowPlanner(UpdateMode).Build([]planRow{{ID: "x"}, {ID: "x"}}, nil) + if err == nil { + t.Fatal("Build accepted duplicate previous keys") + } +} + +func TestPlannerRejectsEmptyKey(t *testing.T) { + _, err := rowPlanner(UpdateMode).Build(nil, []planRow{{ID: ""}}) + if err == nil { + t.Fatal("Build accepted an empty key") + } +} + +func TestRedactedActionsHideKeys(t *testing.T) { + plan, err := rowPlanner(UpdateMode).Build(nil, []planRow{{ID: "private", Value: "value"}}) + if err != nil { + t.Fatal(err) + } + got := plan.RedactedActions() + if len(got) != 1 || got[0].KeyDigest == "private" || got[0].KeyDigest == "" { + t.Fatalf("unexpected redacted action: %#v", got) + } +} + +func TestAuditRecordsDigestRowsSafely(t *testing.T) { + previous := []planRow{{ID: "b", Value: "old"}} + current := []planRow{{ID: "b", Value: "new"}, {ID: "d", Value: "added"}} + plan, err := rowPlanner(SyncMode).Build(previous, current) + if err != nil { + t.Fatal(err) + } + records := plan.AuditRecords() + if len(records) != 2 { + t.Fatalf("records = %d, want 2", len(records)) + } + for _, record := range records { + if record.KeyDigest == "" || record.KeyDigest == "b" || record.KeyDigest == "d" { + t.Fatalf("record exposes raw key: %#v", record) + } + } + update := records[0] + if update.Kind != UpdateAction || update.BeforeDigest == "" || update.AfterDigest == "" || update.BeforeDigest == update.AfterDigest { + t.Fatalf("update record = %#v", update) + } + if len(update.ChangedFields) != 1 || update.ChangedFields[0] != "value" { + t.Fatalf("changed fields = %v", update.ChangedFields) + } + insert := records[1] + if insert.Kind != InsertAction || insert.BeforeDigest != "" || insert.AfterDigest == "" { + t.Fatalf("insert record = %#v", insert) + } +} + +type planFetcher struct { + rows []planRow + init error + fetch error +} + +func (f planFetcher) Init(context.Context) ([]Message, error) { return nil, f.init } +func (f planFetcher) Fetch(context.Context) ([]planRow, error) { + if f.fetch != nil { + return nil, f.fetch + } + return f.rows, nil +} + +type planUpdater struct { + rows []planRow + init error + fetch error + applyErr error + applied []Plan[planRow] + lastPlan Plan[planRow] + updateCall int +} + +func (u *planUpdater) Init(context.Context) ([]Message, error) { return nil, u.init } +func (u *planUpdater) Fetch(context.Context) ([]planRow, error) { + if u.fetch != nil { + return nil, u.fetch + } + return u.rows, nil +} +func (u *planUpdater) Update(_ context.Context, plan Plan[planRow]) error { + u.updateCall++ + u.applied = append(u.applied, plan) + u.lastPlan = plan + return u.applyErr +} + +func TestBuildMergedPlanPreservesDestinationRows(t *testing.T) { + updater := &planUpdater{rows: []planRow{{ID: "keep", Value: "old"}}} + result, err := BuildMergedPlan( + context.Background(), + planFetcher{rows: []planRow{{ID: "new", Value: "new"}}}, + updater, + rowPlanner(UpdateMode), + func(previous, source []planRow) ([]planRow, error) { + return append(append([]planRow(nil), previous...), source...), nil + }, + ) + if err != nil { + t.Fatal(err) + } + if got := result.Plan.Count().Inserted; got != 1 { + t.Fatalf("inserted = %d, want 1", got) + } +} + +func TestBuildPlanDoesNotApply(t *testing.T) { + updater := &planUpdater{rows: []planRow{{ID: "old"}}} + result, err := BuildPlan(context.Background(), planFetcher{rows: []planRow{{ID: "new"}}}, updater, rowPlanner(UpdateMode)) + if err != nil { + t.Fatal(err) + } + if len(updater.applied) != 0 { + t.Fatal("BuildPlan applied changes") + } + if !reflect.DeepEqual(result.Plan.Final, []planRow{{ID: "new"}, {ID: "old"}}) { + t.Fatalf("unexpected final plan: %#v", result.Plan.Final) + } +} + +func TestPlannerRejectsEmptySource(t *testing.T) { + _, err := rowPlanner(SyncMode).Build([]planRow{{ID: "existing"}}, nil) + if err == nil { + t.Fatal("Build accepted an empty source") + } +} + +func TestRunStoreAppliesPlanOnce(t *testing.T) { + output := &planUpdater{rows: []planRow{{ID: "old", Value: "one"}}} + result, err := RunStore(context.Background(), planFetcher{rows: []planRow{{ID: "new", Value: "two"}}}, output, rowPlanner(SyncMode)) + if err != nil { + t.Fatal(err) + } + if len(output.applied) != 1 { + t.Fatalf("applied %d plans, want 1", len(output.applied)) + } + if got := result.Plan.Count(); got.Inserted != 1 || got.Deleted != 1 { + t.Fatalf("counts = %#v", got) + } + // Re-run: destination now matches source, zero actions, no second apply needed. + output.rows = []planRow{{ID: "new", Value: "two"}} + result2, err := RunStore(context.Background(), planFetcher{rows: []planRow{{ID: "new", Value: "two"}}}, output, rowPlanner(SyncMode)) + if err != nil { + t.Fatal(err) + } + if got := result2.Plan.Count(); got != (Counts{}) { + t.Fatalf("second run counts = %#v, want none", got) + } +} + +func TestRunStoreRejectsEmptySourceWithRows(t *testing.T) { + output := &planUpdater{rows: []planRow{{ID: "existing"}}} + _, err := RunStore(context.Background(), planFetcher{rows: nil}, output, rowPlanner(SyncMode)) + if err == nil { + t.Fatal("RunStore accepted an empty source over a populated destination") + } + if output.updateCall != 0 { + t.Fatal("destination was modified before the empty source was rejected") + } +} + +func TestRunStoreEmptyBothSidesIsNoOp(t *testing.T) { + output := &planUpdater{} + result, err := RunStore(context.Background(), planFetcher{rows: nil}, output, rowPlanner(SyncMode)) + if err != nil { + t.Fatal(err) + } + if output.updateCall != 0 || len(result.Plan.Actions) != 0 { + t.Fatalf("expected no-op, got call=%d actions=%d", output.updateCall, len(result.Plan.Actions)) + } +} + +func TestRunStoreFetchErrorLeavesDestinationUntouched(t *testing.T) { + output := &planUpdater{rows: []planRow{{ID: "keep"}}} + boom := errors.New("source unavailable") + _, err := RunStore(context.Background(), planFetcher{rows: nil, fetch: boom}, output, rowPlanner(SyncMode)) + if !errors.Is(err, boom) { + t.Fatalf("err = %v, want %v", err, boom) + } + if output.updateCall != 0 { + t.Fatal("destination was modified after a source fetch error") + } + _, err = RunStore(context.Background(), planFetcher{rows: nil}, &planUpdater{fetch: boom}, rowPlanner(SyncMode)) + if err == nil { + t.Fatal("RunStore accepted a destination fetch error") + } +} + +type invalidFetcher struct { + rows []planRow + invalid int + invalidRows []int +} + +func (f invalidFetcher) Init(context.Context) ([]Message, error) { return nil, nil } +func (f invalidFetcher) Fetch(context.Context) ([]planRow, error) { return f.rows, nil } +func (f invalidFetcher) InvalidCount() int { return f.invalid } +func (f invalidFetcher) InvalidRows() []int { return f.invalidRows } + +func TestRunStoreReportsInvalidRows(t *testing.T) { + result, err := RunStore(context.Background(), + invalidFetcher{rows: []planRow{{ID: "a"}}, invalid: 2, invalidRows: []int{3, 7}}, + &planUpdater{}, rowPlanner(UpdateMode)) + if err != nil { + t.Fatal(err) + } + if result.Invalid != 2 || len(result.InvalidRows) != 2 || result.InvalidRows[0] != 3 { + t.Fatalf("invalid = %d %v", result.Invalid, result.InvalidRows) + } +} + +func TestRunStoreRetryAfterFailedApplyIsSafe(t *testing.T) { + boom := errors.New("apply failed") + // First attempt: apply fails midway; destination is untouched. + first := &planUpdater{rows: []planRow{{ID: "old"}}, applyErr: boom} + _, err := RunStore(context.Background(), planFetcher{rows: []planRow{{ID: "new"}}}, first, rowPlanner(SyncMode)) + if !errors.Is(err, boom) { + t.Fatalf("err = %v, want %v", err, boom) + } + // Retry: a fresh run from the same source produces the identical plan and + // applies successfully to the same final state. + second := &planUpdater{rows: []planRow{{ID: "old"}}} + result, err := RunStore(context.Background(), planFetcher{rows: []planRow{{ID: "new"}}}, second, rowPlanner(SyncMode)) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(result.Plan.Final, []planRow{{ID: "new"}}) { + t.Fatalf("retry final = %#v", result.Plan.Final) + } + if len(second.applied) != 1 { + t.Fatalf("retry applied %d plans", len(second.applied)) + } +} + +func TestRunMergedStoreForcesUpdateMode(t *testing.T) { + // Even a SyncMode planner must not delete through a merged pipeline. + output := &planUpdater{rows: []planRow{{ID: "dest-only"}}} + result, err := RunMergedStore(context.Background(), + planFetcher{rows: []planRow{{ID: "src-only"}}}, + output, + func(previous, source []planRow) ([]planRow, error) { + return append(append([]planRow(nil), previous...), source...), nil + }, + rowPlanner(SyncMode)) + if err != nil { + t.Fatal(err) + } + if got := result.Plan.Count().Deleted; got != 0 { + t.Fatalf("merged pipeline deleted %d rows", got) + } + if result.Plan.Mode != UpdateMode { + t.Fatalf("mode = %v, want UpdateMode", result.Plan.Mode) + } + if output.lastPlan.Mode != UpdateMode { + t.Fatalf("applied plan mode = %v, want UpdateMode", output.lastPlan.Mode) + } +} + +func TestRunMergedStoreEmptySourceKeepsDestination(t *testing.T) { + output := &planUpdater{rows: []planRow{{ID: "keep"}}} + result, err := RunMergedStore(context.Background(), + planFetcher{rows: nil}, + output, + func(previous, source []planRow) ([]planRow, error) { + return append(append([]planRow(nil), previous...), source...), nil + }, + rowPlanner(SyncMode)) + if err != nil { + t.Fatal(err) + } + if got := result.Plan.Count(); got != (Counts{}) { + t.Fatalf("counts = %#v, want none", got) + } +} diff --git a/database/rowsync/store.go b/database/rowsync/store.go new file mode 100644 index 0000000..0ed73de --- /dev/null +++ b/database/rowsync/store.go @@ -0,0 +1,52 @@ +// Package rowsync defines common row storage adapter interfaces and types. +// +// Every durable collection flows through one pipeline: fetch source and +// destination rows, build one rowsync.Plan with a rowsync.Planner, then apply +// that plan to a destination. Destinations never recompute changes: they +// apply Plan.Actions (complete rows in Before/After) and, for PostgreSQL, +// commit base-table and log changes in one transaction. +package rowsync + +import "context" + +// Message is an informational message returned during Init. +type Message struct { + String string `json:"string"` + Type string `json:"type"` +} + +// Fetcher reads records from a source. +type Fetcher[T any] interface { + Init(context.Context) ([]Message, error) + Fetch(context.Context) ([]T, error) +} + +// InvalidReporter reports the count of invalid rows in a source. +type InvalidReporter interface { + InvalidCount() int +} + +// InvalidRowReporter reports the row numbers of invalid rows in a source. +type InvalidRowReporter interface { + InvalidRows() []int +} + +// Updater reads and writes records to a destination. Update applies a plan +// built elsewhere; it must not recompute changes or refetch rows it already +// has (Plan.Before carries the complete previous rows for deletes). +type Updater[T any] interface { + Fetcher[T] + Update(context.Context, Plan[T]) error +} + +// Merge combines a partial source with existing state. Merged pipelines are +// insert-update only: destination rows the source lacks survive. +type Merge[T any] func(previous, current []T) ([]T, error) + +// Result reports the outcome of a fetch → plan → apply run. +type Result[T any] struct { + Source []T + Plan Plan[T] + Invalid int + InvalidRows []int +} diff --git a/database/sheetsdb/fetcher.go b/database/sheetsdb/fetcher.go new file mode 100644 index 0000000..56f7b47 --- /dev/null +++ b/database/sheetsdb/fetcher.go @@ -0,0 +1,414 @@ +// Package sheetsdb provides generic Google Sheets fetchers backed by a +// config-indexed workbook pattern. It does not provide an Updater: Sheets +// are read-only sources. +package sheetsdb + +import ( + "context" + "encoding/csv" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/therootcompany/golib/database/rowsync" + + "github.com/jszwec/csvutil" + "github.com/therootcompany/golib/io/transform/gsheet2csv" + + "github.com/therootcompany/golib/database/csvx" +) + +const DefaultTimeout = 30 * time.Second +const MaxSourceBytes = 16 << 20 + +var ErrSourceNotFound = errors.New("source not found") +var ErrSourceNotConfigured = errors.New("source not configured") + +// ConfigRow is the required shape of the workbook's index tab. +type ConfigRow struct { + Description string `json:"description"` + Key string `json:"key"` + Value string `json:"value"` +} + +// Config is the typed config-tab result. Rows stay ordered so duplicate keys +// can be detected instead of silently resolved by map iteration. +type Config struct { + Rows []ConfigRow +} + +// SourceURL returns the configured URL/path for a source key. +func (c Config) SourceURL(key string) (string, bool, bool) { + wanted := NormalizeConfigKey(key) + var value string + found := false + duplicate := false + for _, row := range c.Rows { + if NormalizeConfigKey(row.Key) != wanted { + continue + } + if found { + duplicate = true + continue + } + value = strings.TrimSpace(row.Value) + found = true + } + return value, found, duplicate +} + +// Environment returns a normalized environment value. Both NAME and env:NAME +// forms are accepted. +func (c Config) Environment(name string) (string, bool, bool) { + return c.value(NormalizeEnvironmentKey(name)) +} + +// EnvironmentValues returns ALL_CAPS and env:-prefixed config values. +func (c Config) EnvironmentValues() (map[string]string, error) { + values := make(map[string]string) + for _, row := range c.Rows { + rawKey := strings.TrimSpace(row.Key) + key := rawKey + if strings.HasPrefix(strings.ToLower(key), "env:") { + key = strings.TrimSpace(key[len("env:"):]) + } else if key != strings.ToUpper(key) { + continue + } + key = strings.ToUpper(key) + if key == "" { + return nil, fmt.Errorf("empty environment key") + } + if _, duplicate := values[key]; duplicate { + return nil, fmt.Errorf("duplicate environment key %q", key) + } + values[key] = strings.TrimSpace(row.Value) + } + return values, nil +} + +func (c Config) value(wanted string) (string, bool, bool) { + var value string + found := false + duplicate := false + for _, row := range c.Rows { + if NormalizeEnvironmentKey(row.Key) != wanted { + continue + } + if found { + duplicate = true + continue + } + value = strings.TrimSpace(row.Value) + found = true + } + return value, found, duplicate +} + +// NormalizeConfigKey lowercases, trims, and applies collection aliases. +func NormalizeConfigKey(value string) string { + value = strings.ToLower(strings.TrimSpace(value)) + if value == "profiles" { + return "identities" + } + return value +} + +// NormalizeEnvironmentKey normalizes an environment key by lowercasing, +// trimming, and stripping an optional env: prefix. +func NormalizeEnvironmentKey(value string) string { + value = NormalizeConfigKey(value) + value = strings.TrimPrefix(value, "env:") + return value +} + +// ConfigCache provides thread-safe, once-only loading of a workbook's config tab. +type ConfigCache struct { + mu sync.Mutex + rows []ConfigRow + loaded bool + loading bool + finished chan struct{} +} + +// Load fetches the config tab if not already loaded. +func (c *ConfigCache) Load(ctx context.Context, indexURL string, client *http.Client) (Config, error) { + for { + c.mu.Lock() + if c.loaded { + rows := append([]ConfigRow(nil), c.rows...) + c.mu.Unlock() + return Config{Rows: rows}, nil + } + if c.loading { + finished := c.finished + c.mu.Unlock() + select { + case <-finished: + continue + case <-ctx.Done(): + return Config{}, ctx.Err() + } + } + c.loading = true + c.finished = make(chan struct{}) + finished := c.finished + c.mu.Unlock() + + rows, err := (&WebFetcher[ConfigRow]{URL: indexURL, Client: client}).Fetch(ctx) + c.mu.Lock() + c.loading = false + if err == nil { + c.rows = append([]ConfigRow(nil), rows...) + c.loaded = true + } + close(finished) + c.mu.Unlock() + if err != nil { + return Config{}, fmt.Errorf("fetch Google Sheets config: %w", err) + } + return Config{Rows: append([]ConfigRow(nil), rows...)}, nil + } +} + +var configCaches sync.Map + +func sharedConfigCache(indexURL string) *ConfigCache { + if cached, ok := configCaches.Load(indexURL); ok { + return cached.(*ConfigCache) + } + cache := &ConfigCache{} + actual, _ := configCaches.LoadOrStore(indexURL, cache) + return actual.(*ConfigCache) +} + +// LoadConfig fetches and caches the workbook config tab using a process-wide cache. +func LoadConfig(ctx context.Context, indexURL string, client *http.Client) (Config, error) { + if strings.TrimSpace(indexURL) == "" { + return Config{}, errors.New("a Google Sheets index URL is required") + } + return sharedConfigCache(indexURL).Load(ctx, indexURL, client) +} + +// WebFetcher fetches one Google Sheets tab (or local CSV/TSV file) and +// decodes it into T. T should use json tags, and nullable source cells +// should use csvx fields. +type WebFetcher[T any] struct { + URL string + Format string + Client *http.Client +} + +func (f *WebFetcher[T]) Init(ctx context.Context) ([]rowsync.Message, error) { + if f == nil || strings.TrimSpace(f.URL) == "" { + return nil, errors.New("source URL or filepath is required") + } + if isLocalPath(f.URL) { + if _, err := os.Stat(f.URL); err != nil { + return nil, fmt.Errorf("stat source %q: %w", f.URL, err) + } + return []rowsync.Message{{String: filepath.Base(f.URL), Type: "debug"}}, nil + } + csvURL, err := sheetCSVURL(f.URL) + if err != nil { + return nil, err + } + client := f.client() + req, err := http.NewRequestWithContext(ctx, http.MethodHead, csvURL, nil) + if err != nil { + return nil, fmt.Errorf("create Google Sheets request: %w", err) + } + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("connect to Google Sheets: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("fetch from Google Sheets returned HTTP %d", resp.StatusCode) + } + return []rowsync.Message{{String: "Google Sheets", Type: "debug"}}, nil +} + +func (f *WebFetcher[T]) Fetch(ctx context.Context) ([]T, error) { + if f == nil || strings.TrimSpace(f.URL) == "" { + return nil, errors.New("a Google Sheets URL is required") + } + body, err := f.get(ctx) + if err != nil { + return nil, err + } + defer func() { _ = body.Close() }() + + format := strings.ToLower(strings.TrimSpace(f.Format)) + if format == "" { + format = detectFormat(f.URL) + } + if format != "csv" && format != "tsv" && format != "tab" { + return nil, fmt.Errorf("unsupported Google Sheets format %q", f.Format) + } + limited := &io.LimitedReader{R: body, N: MaxSourceBytes + 1} + var rows []T + if isLocalPath(f.URL) { + reader := csv.NewReader(limited) + reader.Comment = '#' + if format == "tsv" || format == "tab" { + reader.Comma = '\t' + } + rows, err = parseSourceCSV[T](reader) + } else { + reader := gsheet2csv.NewReader(limited) + rows, err = parseSourceCSV[T](reader) + } + if err != nil { + return nil, fmt.Errorf("parse Google Sheets rows from %q: %w", f.URL, err) + } + if limited.N == 0 { + return nil, fmt.Errorf("the Google Sheets source exceeds %d bytes", MaxSourceBytes) + } + return rows, nil +} + +// parseSourceCSV reads human-edited source data. Known tagged columns are +// mapped and extra columns are ignored so harmless additions do not break a +// refresh. +func parseSourceCSV[T any](reader csvutil.Reader) ([]T, error) { + return csvx.ParseCSV[T](reader) +} + +func (f *WebFetcher[T]) get(ctx context.Context) (io.ReadCloser, error) { + if isLocalPath(f.URL) { + return os.Open(f.URL) + } + csvURL, err := sheetCSVURL(f.URL) + if err != nil { + return nil, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, csvURL, nil) + if err != nil { + return nil, fmt.Errorf("create Google Sheets request: %w", err) + } + resp, err := f.client().Do(req) + if err != nil { + return nil, fmt.Errorf("fetch Google Sheets: %w", err) + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + _ = resp.Body.Close() + return nil, fmt.Errorf("fetch from Google Sheets returned HTTP %d", resp.StatusCode) + } + return resp.Body, nil +} + +func (f *WebFetcher[T]) client() *http.Client { + if f.Client != nil { + return f.Client + } + return &http.Client{Timeout: DefaultTimeout} +} + +func isLocalPath(value string) bool { + return !strings.HasPrefix(strings.ToLower(strings.TrimSpace(value)), "http://") && + !strings.HasPrefix(strings.ToLower(strings.TrimSpace(value)), "https://") +} + +func detectFormat(value string) string { + if strings.HasSuffix(strings.ToLower(value), ".tsv") || strings.HasSuffix(strings.ToLower(value), ".tab") { + return "tsv" + } + return "csv" +} + +func sheetCSVURL(rawURL string) (string, error) { + u, err := url.Parse(rawURL) + if err != nil || u.Scheme == "" || u.Host == "" { + return "", fmt.Errorf("invalid Google Sheets URL") + } + if !strings.Contains(u.Path, "/spreadsheets/d/") { + return "", fmt.Errorf("URL is not a Google Sheets URL") + } + docID, gid := gsheet2csv.ParseIDs(rawURL) + if docID == "" { + return "", fmt.Errorf("the Google Sheets document ID is missing") + } + return gsheet2csv.ToCSVURL(docID, gid), nil +} + +// IndexedFetcher resolves a source tab URL from the workbook's config tab, +// then fetches and decodes that tab into T. +type IndexedFetcher[T any] struct { + IndexURL string + Key string + Client *http.Client + Config *ConfigCache +} + +func (f *IndexedFetcher[T]) Init(ctx context.Context) ([]rowsync.Message, error) { + if f == nil || strings.TrimSpace(f.IndexURL) == "" { + return nil, errors.New("a Google Sheets index URL is required") + } + if strings.TrimSpace(f.Key) == "" { + return nil, errors.New("a Google Sheets source key is required") + } + cache := f.Config + if cache == nil { + cache = sharedConfigCache(f.IndexURL) + } + if _, err := cache.Load(ctx, f.IndexURL, f.Client); err != nil { + return nil, fmt.Errorf("initialize Google Sheets index: %w", err) + } + return []rowsync.Message{{String: "Google Sheets index", Type: "debug"}}, nil +} + +func (f *IndexedFetcher[T]) Fetch(ctx context.Context) ([]T, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + cache := f.Config + if cache == nil { + cache = sharedConfigCache(f.IndexURL) + } + configValue, err := cache.Load(ctx, f.IndexURL, f.Client) + if err != nil { + return nil, err + } + value, found, duplicate := configValue.SourceURL(f.Key) + if duplicate { + return nil, fmt.Errorf("the Google Sheets source %q is configured more than once", f.Key) + } + if !found { + return nil, fmt.Errorf("%w: Google Sheets source %q", ErrSourceNotFound, f.Key) + } + if value != "" && isLocalPath(f.IndexURL) && isLocalPath(value) && !filepath.IsAbs(value) { + value = filepath.Join(filepath.Dir(f.IndexURL), value) + } + if value == "" { + return nil, fmt.Errorf("%w: Google Sheets source %q has an empty URL", ErrSourceNotConfigured, f.Key) + } + return (&WebFetcher[T]{URL: value, Client: f.Client}).Fetch(ctx) +} + +// OptionalIndexedFetcher treats a missing config source as an empty source. +type OptionalIndexedFetcher[T any] struct{ Indexed IndexedFetcher[T] } + +func (f *OptionalIndexedFetcher[T]) Init(ctx context.Context) ([]rowsync.Message, error) { + if f == nil { + return nil, errors.New("optional source is required") + } + return f.Indexed.Init(ctx) +} + +func (f *OptionalIndexedFetcher[T]) Fetch(ctx context.Context) ([]T, error) { + if f == nil { + return nil, errors.New("optional source is required") + } + rows, err := f.Indexed.Fetch(ctx) + if errors.Is(err, ErrSourceNotFound) || errors.Is(err, ErrSourceNotConfigured) { + return nil, nil + } + return rows, err +} diff --git a/database/sheetsdb/fetcher_test.go b/database/sheetsdb/fetcher_test.go new file mode 100644 index 0000000..2e58f7a --- /dev/null +++ b/database/sheetsdb/fetcher_test.go @@ -0,0 +1,21 @@ +package sheetsdb + +import "testing" + +func TestConfigSourceURLIdentitiesPrimary(t *testing.T) { + config := Config{Rows: []ConfigRow{{Key: "identities", Value: "identities.tsv"}}} + + value, found, duplicate := config.SourceURL("identities") + if !found || duplicate || value != "identities.tsv" { + t.Fatalf("SourceURL(identities) = %q, %v, %v; want identities.tsv, true, false", value, found, duplicate) + } +} + +func TestConfigSourceURLProfilesAlias(t *testing.T) { + config := Config{Rows: []ConfigRow{{Key: "identities", Value: "identities.tsv"}}} + + value, found, duplicate := config.SourceURL("profiles") + if !found || duplicate || value != "identities.tsv" { + t.Fatalf("SourceURL(profiles) = %q, %v, %v; want identities.tsv, true, false", value, found, duplicate) + } +}