From 1796b5b872a06bcc22e30b92f04257f25f1933e7 Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Tue, 14 Jul 2026 11:21:38 +0800 Subject: [PATCH] storage: add upload_files table and UploadRepo Add the upload_files SQLite table and UploadRepo CRUD methods (Create/Get/List/Delete). The table stores file_id, name, size, sha256, and uploaded_at as a queryable index. The .meta.json sidecar remains the source of truth for uploads; the DB is only an index for admin visibility. On startup the next commit will backfill any pre-existing sidecars into this table. Tests cover Create/Get/List/Delete round-trip, name search filtering, descending time ordering, and limit defaults/cap. Co-Authored-By: Claude --- internal/storage/sqlite.go | 9 ++ internal/storage/upload_repo.go | 147 ++++++++++++++++++++++++ internal/storage/upload_repo_test.go | 162 +++++++++++++++++++++++++++ 3 files changed, 318 insertions(+) create mode 100644 internal/storage/upload_repo.go create mode 100644 internal/storage/upload_repo_test.go diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go index f0e94be..1808ac4 100644 --- a/internal/storage/sqlite.go +++ b/internal/storage/sqlite.go @@ -38,6 +38,15 @@ CREATE TABLE IF NOT EXISTS audit_log ( details TEXT ); CREATE INDEX IF NOT EXISTS idx_audit_ts ON audit_log(timestamp DESC); + +CREATE TABLE IF NOT EXISTS upload_files ( + file_id TEXT PRIMARY KEY, + name TEXT NOT NULL, + size INTEGER NOT NULL, + sha256 TEXT NOT NULL, + uploaded_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_uploads_ts ON upload_files(uploaded_at DESC); ` // DB is the storage handle. diff --git a/internal/storage/upload_repo.go b/internal/storage/upload_repo.go new file mode 100644 index 0000000..700effb --- /dev/null +++ b/internal/storage/upload_repo.go @@ -0,0 +1,147 @@ +package storage + +import ( + "context" + "database/sql" + "errors" + "fmt" + "strings" + "time" +) + +// UploadMeta is the database index record for an uploaded file. +// +// The .meta.json sidecar remains the source of truth; this struct is the +// queryable mirror used by the admin UI. +type UploadMeta struct { + FileID string + Name string + Size int64 + SHA256 string + UploadedAt time.Time +} + +// UploadRepo provides CRUD operations for upload file index records. +type UploadRepo struct { + db *DB +} + +// Uploads returns a repository bound to this DB handle. +func (d *DB) Uploads() *UploadRepo { + return &UploadRepo{db: d} +} + +// Create inserts a new upload index record. Timestamps are stored as Unix +// nanoseconds for stable ORDER BY semantics. +func (r *UploadRepo) Create(ctx context.Context, fileID, name string, size int64, sha256Hex string, uploadedAt time.Time) error { + _, err := r.db.sqlDB.ExecContext(ctx, ` + INSERT INTO upload_files (file_id, name, size, sha256, uploaded_at) + VALUES (?, ?, ?, ?, ?)`, + fileID, name, size, sha256Hex, uploadedAt.UnixNano(), + ) + if err != nil { + if strings.Contains(err.Error(), "UNIQUE constraint failed") { + return fmt.Errorf("storage: create upload %s: primary key conflict: %w", fileID, err) + } + return fmt.Errorf("storage: create upload %s: %w", fileID, err) + } + return nil +} + +// Get retrieves an upload index record by file_id. If the record does not +// exist, it returns ErrNotFound. +func (r *UploadRepo) Get(ctx context.Context, fileID string) (UploadMeta, error) { + row := r.db.sqlDB.QueryRowContext(ctx, ` + SELECT file_id, name, size, sha256, uploaded_at + FROM upload_files + WHERE file_id = ?`, fileID) + + m, err := scanUpload(row.Scan) + if errors.Is(err, sql.ErrNoRows) { + return UploadMeta{}, ErrNotFound + } + if err != nil { + return UploadMeta{}, fmt.Errorf("storage: get upload %s: %w", fileID, err) + } + return m, nil +} + +// List returns upload index records ordered by uploaded_at descending. +// +// If search is non-empty, name is filtered with a case-insensitive LIKE. +// A limit of zero or less falls back to 100; values above 500 are capped. +func (r *UploadRepo) List(ctx context.Context, search string, limit int) ([]UploadMeta, error) { + if limit <= 0 { + limit = 100 + } + if limit > 500 { + limit = 500 + } + + var rows *sql.Rows + var err error + if search != "" { + rows, err = r.db.sqlDB.QueryContext(ctx, ` + SELECT file_id, name, size, sha256, uploaded_at + FROM upload_files + WHERE name LIKE ? + ORDER BY uploaded_at DESC + LIMIT ?`, + "%"+search+"%", limit, + ) + } else { + rows, err = r.db.sqlDB.QueryContext(ctx, ` + SELECT file_id, name, size, sha256, uploaded_at + FROM upload_files + ORDER BY uploaded_at DESC + LIMIT ?`, limit) + } + if err != nil { + return nil, fmt.Errorf("storage: list uploads: %w", err) + } + defer rows.Close() + + var out []UploadMeta + for rows.Next() { + m, err := scanUpload(rows.Scan) + if err != nil { + return nil, fmt.Errorf("storage: list uploads: %w", err) + } + out = append(out, m) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("storage: list uploads: %w", err) + } + if out == nil { + // Force empty slice (not null) in JSON. + out = []UploadMeta{} + } + return out, nil +} + +// Delete removes an upload index record by file_id. If the record does not +// exist, it returns ErrNotFound. +func (r *UploadRepo) Delete(ctx context.Context, fileID string) error { + res, err := r.db.sqlDB.ExecContext(ctx, `DELETE FROM upload_files WHERE file_id = ?`, fileID) + if err != nil { + return fmt.Errorf("storage: delete upload %s: %w", fileID, err) + } + n, err := res.RowsAffected() + if err != nil { + return fmt.Errorf("storage: delete upload %s: rows affected: %w", fileID, err) + } + if n == 0 { + return ErrNotFound + } + return nil +} + +func scanUpload(scan scanFunc) (UploadMeta, error) { + var m UploadMeta + var uploadedAt int64 + if err := scan(&m.FileID, &m.Name, &m.Size, &m.SHA256, &uploadedAt); err != nil { + return UploadMeta{}, err + } + m.UploadedAt = time.Unix(0, uploadedAt) + return m, nil +} diff --git a/internal/storage/upload_repo_test.go b/internal/storage/upload_repo_test.go new file mode 100644 index 0000000..99f1049 --- /dev/null +++ b/internal/storage/upload_repo_test.go @@ -0,0 +1,162 @@ +package storage + +import ( + "context" + "errors" + "fmt" + "testing" + "time" +) + +func newUploadRepo(t *testing.T) *UploadRepo { + t.Helper() + db, err := Open(":memory:") + if err != nil { + t.Fatalf("Open in-memory db: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + return db.Uploads() +} + +func assertUploadEqual(t *testing.T, got, want UploadMeta) { + t.Helper() + if got.FileID != want.FileID { + t.Errorf("FileID: got %q, want %q", got.FileID, want.FileID) + } + if got.Name != want.Name { + t.Errorf("Name: got %q, want %q", got.Name, want.Name) + } + if got.Size != want.Size { + t.Errorf("Size: got %d, want %d", got.Size, want.Size) + } + if got.SHA256 != want.SHA256 { + t.Errorf("SHA256: got %q, want %q", got.SHA256, want.SHA256) + } + if !got.UploadedAt.Equal(want.UploadedAt) { + t.Errorf("UploadedAt: got %v, want %v", got.UploadedAt, want.UploadedAt) + } +} + +func TestUploadRepo_CreateGetListDelete(t *testing.T) { + repo := newUploadRepo(t) + ctx := context.Background() + uploadedAt := time.Unix(0, time.Now().UnixNano()) + want := UploadMeta{ + FileID: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + Name: "report.csv", + Size: 42, + SHA256: "deadbeef", + UploadedAt: uploadedAt, + } + if err := repo.Create(ctx, want.FileID, want.Name, want.Size, want.SHA256, want.UploadedAt); err != nil { + t.Fatalf("Create: %v", err) + } + got, err := repo.Get(ctx, want.FileID) + if err != nil { + t.Fatalf("Get: %v", err) + } + assertUploadEqual(t, got, want) + if err := repo.Create(ctx, want.FileID, "other", 1, "abcd", time.Now()); err == nil { + t.Errorf("duplicate Create succeeded, want error") + } + list, err := repo.List(ctx, "", 100) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(list) != 1 { + t.Fatalf("List len: got %d, want 1", len(list)) + } + assertUploadEqual(t, list[0], want) + if err := repo.Delete(ctx, want.FileID); err != nil { + t.Fatalf("Delete: %v", err) + } + if _, err := repo.Get(ctx, want.FileID); !errors.Is(err, ErrNotFound) { + t.Fatalf("Get after Delete: got %v, want ErrNotFound", err) + } + if err := repo.Delete(ctx, want.FileID); !errors.Is(err, ErrNotFound) { + t.Fatalf("Delete again: got %v, want ErrNotFound", err) + } +} + +func TestUploadRepo_ListOrderAndSearch(t *testing.T) { + repo := newUploadRepo(t) + ctx := context.Background() + base := time.Unix(0, time.Now().UnixNano()) + records := []UploadMeta{ + {FileID: "00000000000000000000000000000001", Name: "alpha.csv", Size: 1, SHA256: "a", UploadedAt: base.Add(-2 * time.Hour)}, + {FileID: "00000000000000000000000000000002", Name: "beta.log", Size: 2, SHA256: "b", UploadedAt: base.Add(-1 * time.Hour)}, + {FileID: "00000000000000000000000000000003", Name: "gamma.csv", Size: 3, SHA256: "c", UploadedAt: base}, + } + for _, r := range records { + if err := repo.Create(ctx, r.FileID, r.Name, r.Size, r.SHA256, r.UploadedAt); err != nil { + t.Fatalf("Create %s: %v", r.FileID, err) + } + } + all, err := repo.List(ctx, "", 100) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(all) != 3 { + t.Fatalf("List len: got %d, want 3", len(all)) + } + wantOrder := []string{"00000000000000000000000000000003", "00000000000000000000000000000002", "00000000000000000000000000000001"} + for i, id := range wantOrder { + if all[i].FileID != id { + t.Errorf("List order[%d]: got %q, want %q", i, all[i].FileID, id) + } + } + csv, err := repo.List(ctx, "csv", 100) + if err != nil { + t.Fatalf("List search: %v", err) + } + if len(csv) != 2 { + t.Fatalf("search csv len: got %d, want 2", len(csv)) + } + if csv[0].Name != "gamma.csv" || csv[1].Name != "alpha.csv" { + t.Errorf("search csv order: got %v", []string{csv[0].Name, csv[1].Name}) + } + beta, err := repo.List(ctx, "beta", 100) + if err != nil { + t.Fatalf("List search beta: %v", err) + } + if len(beta) != 1 || beta[0].Name != "beta.log" { + t.Errorf("search beta: got %+v", beta) + } +} + +func TestUploadRepo_ListLimitCap(t *testing.T) { + repo := newUploadRepo(t) + ctx := context.Background() + base := time.Unix(0, time.Now().UnixNano()) + for i := 0; i < 10; i++ { + id := fmt.Sprintf("%032d", i) + if err := repo.Create(ctx, id, "x", int64(i), "h", base.Add(time.Duration(i)*time.Second)); err != nil { + t.Fatalf("Create %d: %v", i, err) + } + } + list, err := repo.List(ctx, "", 0) + if err != nil { + t.Fatalf("List default limit: %v", err) + } + // Default limit is 100 but only 10 rows exist. + if len(list) != 10 { + t.Errorf("default limit: got %d, want 10", len(list)) + } + + list, err = repo.List(ctx, "", 501) + if err != nil { + t.Fatalf("List cap: %v", err) + } + if len(list) != 10 { + t.Errorf("cap limit: got %d, want 10", len(list)) + } + + // Explicit small limit is respected. + list, err = repo.List(ctx, "", 3) + if err != nil { + t.Fatalf("List small limit: %v", err) + } + if len(list) != 3 { + t.Errorf("small limit: got %d, want 3", len(list)) + } +}