feat: add database/{csvx,rowsync,csvdb,pgdb,sheetsdb}

This commit is contained in:
AJ ONeal 2026-09-16 15:42:05 -06:00
parent 04aaadff75
commit ca6e271f66
No known key found for this signature in database
35 changed files with 4295 additions and 0 deletions

45
database/csvdb/atomic.go Normal file
View File

@ -0,0 +1,45 @@
package csvdb
import (
"crypto/rand"
"encoding/base64"
"fmt"
"os"
)
// createTemp creates a temporary file within root using the pattern
// <name>.<rnd>.tmp, where <rnd> 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)
}

304
database/csvdb/file.go Normal file
View File

@ -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()
}

View File

@ -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")
}
}

145
database/csvdb/file_test.go Normal file
View File

@ -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))
}
}

34
database/csvdb/gate.go Normal file
View File

@ -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()

View File

@ -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"`
}

View File

@ -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)
}
}

100
database/csvx/bool.go Normal file
View File

@ -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
}

68
database/csvx/boolish.go Normal file
View File

@ -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 }

108
database/csvx/bytes.go Normal file
View File

@ -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
}

76
database/csvx/csv.go Normal file
View File

@ -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
}),
)
}

229
database/csvx/csvx_test.go Normal file
View File

@ -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)
}
}

112
database/csvx/date.go Normal file
View File

@ -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
}

175
database/csvx/delimited.go Normal file
View File

@ -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
}

View File

@ -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)
}
}

9
database/csvx/doc.go Normal file
View File

@ -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

172
database/csvx/encoding.go Normal file
View File

@ -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
}

View File

@ -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)
}
}

130
database/csvx/http.go Normal file
View File

@ -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
}

View File

@ -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"
}
}

156
database/csvx/number.go Normal file
View File

@ -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
}

42
database/csvx/rawjson.go Normal file
View File

@ -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) }

89
database/csvx/text.go Normal file
View File

@ -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
}

120
database/csvx/timestamp.go Normal file
View File

@ -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)
}

View File

@ -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
}

127
database/csvx/uuid.go Normal file
View File

@ -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
}

19
database/go.mod Normal file
View File

@ -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
)

38
database/go.sum Normal file
View File

@ -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=

172
database/pgdb/conn.go Normal file
View File

@ -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)
}

View File

@ -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
}

318
database/rowsync/plan.go Normal file
View File

@ -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
}

View File

@ -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: &current[1], ChangedFields: []string{"value"}},
{Kind: InsertAction, Key: "d", After: &current[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)
}
}

52
database/rowsync/store.go Normal file
View File

@ -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
}

View File

@ -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
}

View File

@ -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)
}
}