Make artifact and media metadata first-class server records
Artifact and media metadata were previously reconstructed from job result refs and JSON sidecar files. This change regularizes metadata into SQLite, keeps filesystem blobs in place, and preserves backward compatibility via lazy fallback and backfill from existing job refs and JSON metadata. Constraint: Blob storage remains on the local filesystem in this phase Rejected: Migrate blobs into SQLite | larger scope and worse operational profile for current media sizes Rejected: Hard cutover without fallback | unsafe for historical data already on the test server Confidence: medium Scope-risk: moderate Directive: Treat SQLite as the metadata source of truth; JSON sidecars are compatibility fallback only Tested: go test ./...; deployed to test server 101.42.99.35; verified /v1/artifacts, /v1/artifacts/:id, signed media URL 200, unsigned content 401 Not-tested: Full historical backfill sweep over all existing artifact rows under production-sized data volume
This commit is contained in:
wtgoku
@@ -85,6 +85,50 @@ func (r *sqliteRepository) migrate() error {
|
||||
`CREATE INDEX IF NOT EXISTS idx_jobs_status_created_at ON jobs(status, created_at DESC);`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_jobs_project_created_at ON jobs(project_id, created_at DESC);`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_jobs_skill_created_at ON jobs(skill_id, created_at DESC);`,
|
||||
`CREATE TABLE IF NOT EXISTS media_records (
|
||||
media_id TEXT PRIMARY KEY,
|
||||
user_id TEXT NOT NULL,
|
||||
artifact_id TEXT,
|
||||
project_id TEXT,
|
||||
filename TEXT NOT NULL,
|
||||
content_type TEXT NOT NULL,
|
||||
size_bytes INTEGER NOT NULL,
|
||||
created_at TEXT NOT NULL,
|
||||
url TEXT NOT NULL,
|
||||
visibility TEXT,
|
||||
sha256 TEXT,
|
||||
local_path TEXT NOT NULL
|
||||
);`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_media_records_user_created_at ON media_records(user_id, created_at DESC);`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_media_records_project_created_at ON media_records(project_id, created_at DESC);`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_media_records_artifact_id ON media_records(artifact_id);`,
|
||||
`CREATE TABLE IF NOT EXISTS artifacts (
|
||||
artifact_id TEXT PRIMARY KEY,
|
||||
user_id TEXT NOT NULL,
|
||||
job_id TEXT NOT NULL,
|
||||
project_id TEXT,
|
||||
result_index INTEGER NOT NULL,
|
||||
media_id TEXT,
|
||||
filename TEXT NOT NULL,
|
||||
content_type TEXT NOT NULL,
|
||||
size_bytes INTEGER NOT NULL,
|
||||
created_at TEXT NOT NULL,
|
||||
expires_at TEXT,
|
||||
visibility TEXT,
|
||||
sha256 TEXT,
|
||||
storage_status TEXT,
|
||||
source_skill_id TEXT,
|
||||
source_model_id TEXT,
|
||||
source_route_key TEXT,
|
||||
source_input_json TEXT,
|
||||
prompt_text TEXT,
|
||||
usage_json TEXT,
|
||||
ref_json TEXT NOT NULL
|
||||
);`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_artifacts_user_created_at ON artifacts(user_id, created_at DESC, artifact_id DESC);`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_artifacts_job_result_index ON artifacts(job_id, result_index ASC);`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_artifacts_project_created_at ON artifacts(project_id, created_at DESC, artifact_id DESC);`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_artifacts_media_id ON artifacts(media_id);`,
|
||||
`CREATE TABLE IF NOT EXISTS skill_routes (
|
||||
route_key TEXT NOT NULL,
|
||||
scope_key TEXT NOT NULL,
|
||||
@@ -474,6 +518,226 @@ func (r *sqliteRepository) UnsetRoute(projectID, routeKey string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *sqliteRepository) UpsertMedia(record mediaRecord) error {
|
||||
_, err := r.db.Exec(
|
||||
`INSERT INTO media_records (
|
||||
media_id, user_id, artifact_id, project_id, filename, content_type,
|
||||
size_bytes, created_at, url, visibility, sha256, local_path
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(media_id) DO UPDATE SET
|
||||
user_id = excluded.user_id,
|
||||
artifact_id = excluded.artifact_id,
|
||||
project_id = excluded.project_id,
|
||||
filename = excluded.filename,
|
||||
content_type = excluded.content_type,
|
||||
size_bytes = excluded.size_bytes,
|
||||
created_at = excluded.created_at,
|
||||
url = excluded.url,
|
||||
visibility = excluded.visibility,
|
||||
sha256 = excluded.sha256,
|
||||
local_path = excluded.local_path`,
|
||||
record.ID,
|
||||
record.UserID,
|
||||
nullableString(record.ArtifactID),
|
||||
nullableString(record.ProjectID),
|
||||
record.Filename,
|
||||
record.ContentType,
|
||||
record.SizeBytes,
|
||||
record.CreatedAt,
|
||||
record.URL,
|
||||
nullableString(record.Visibility),
|
||||
nullableString(record.SHA256),
|
||||
record.LocalPath,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("upsert media: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *sqliteRepository) GetMedia(mediaID string) (*mediaRecord, bool, error) {
|
||||
row := r.db.QueryRow(
|
||||
`SELECT media_id, user_id, artifact_id, project_id, filename, content_type,
|
||||
size_bytes, created_at, url, visibility, sha256, local_path
|
||||
FROM media_records WHERE media_id = ?`,
|
||||
mediaID,
|
||||
)
|
||||
var (
|
||||
record mediaRecord
|
||||
artifactID, projectID sql.NullString
|
||||
visibility, sha256 sql.NullString
|
||||
)
|
||||
err := row.Scan(
|
||||
&record.ID,
|
||||
&record.UserID,
|
||||
&artifactID,
|
||||
&projectID,
|
||||
&record.Filename,
|
||||
&record.ContentType,
|
||||
&record.SizeBytes,
|
||||
&record.CreatedAt,
|
||||
&record.URL,
|
||||
&visibility,
|
||||
&sha256,
|
||||
&record.LocalPath,
|
||||
)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return nil, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("scan media: %w", err)
|
||||
}
|
||||
record.ArtifactID = artifactID.String
|
||||
record.ProjectID = projectID.String
|
||||
record.Visibility = visibility.String
|
||||
record.SHA256 = sha256.String
|
||||
return &record, true, nil
|
||||
}
|
||||
|
||||
func (r *sqliteRepository) UpsertArtifacts(userID, projectID string, items []artifact) error {
|
||||
if strings.TrimSpace(userID) == "" || len(items) == 0 {
|
||||
return nil
|
||||
}
|
||||
tx, err := r.db.Begin()
|
||||
if err != nil {
|
||||
return fmt.Errorf("begin artifacts upsert: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
|
||||
stmt, err := tx.Prepare(
|
||||
`INSERT INTO artifacts (
|
||||
artifact_id, user_id, job_id, project_id, result_index, media_id,
|
||||
filename, content_type, size_bytes, created_at, expires_at,
|
||||
visibility, sha256, storage_status, source_skill_id, source_model_id,
|
||||
source_route_key, source_input_json, prompt_text, usage_json, ref_json
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(artifact_id) DO UPDATE SET
|
||||
user_id = excluded.user_id,
|
||||
job_id = excluded.job_id,
|
||||
project_id = excluded.project_id,
|
||||
result_index = excluded.result_index,
|
||||
media_id = excluded.media_id,
|
||||
filename = excluded.filename,
|
||||
content_type = excluded.content_type,
|
||||
size_bytes = excluded.size_bytes,
|
||||
created_at = excluded.created_at,
|
||||
expires_at = excluded.expires_at,
|
||||
visibility = excluded.visibility,
|
||||
sha256 = excluded.sha256,
|
||||
storage_status = excluded.storage_status,
|
||||
source_skill_id = excluded.source_skill_id,
|
||||
source_model_id = excluded.source_model_id,
|
||||
source_route_key = excluded.source_route_key,
|
||||
source_input_json = excluded.source_input_json,
|
||||
prompt_text = excluded.prompt_text,
|
||||
usage_json = excluded.usage_json,
|
||||
ref_json = excluded.ref_json`,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("prepare artifacts upsert: %w", err)
|
||||
}
|
||||
defer stmt.Close()
|
||||
|
||||
for _, item := range items {
|
||||
_, idx, ok := parseArtifactID(item.ID)
|
||||
if !ok {
|
||||
return fmt.Errorf("invalid artifact id: %s", item.ID)
|
||||
}
|
||||
refJSON, err := marshalJSON(item.Ref)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
sourceInputJSON, err := marshalNullableJSON(item.SourceInput)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
usageJSON, err := marshalNullableJSON(item.Usage)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := stmt.Exec(
|
||||
item.ID,
|
||||
userID,
|
||||
item.JobID,
|
||||
nullableString(defaultString(item.ProjectID, projectID)),
|
||||
idx,
|
||||
nullableString(item.MediaID),
|
||||
item.Filename,
|
||||
item.ContentType,
|
||||
item.SizeBytes,
|
||||
item.CreatedAt,
|
||||
nullableString(item.ExpiresAt),
|
||||
nullableString(item.Visibility),
|
||||
nullableString(item.SHA256),
|
||||
nullableString(item.StorageStatus),
|
||||
nullableString(item.SourceSkillID),
|
||||
nullableString(item.SourceModelID),
|
||||
nullableString(item.SourceRouteKey),
|
||||
sourceInputJSON,
|
||||
nullableString(item.PromptText),
|
||||
usageJSON,
|
||||
refJSON,
|
||||
); err != nil {
|
||||
return fmt.Errorf("exec artifacts upsert: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("commit artifacts upsert: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *sqliteRepository) GetArtifact(userID, artifactID string) (*artifact, resultRef, bool, error) {
|
||||
row := r.db.QueryRow(
|
||||
`SELECT artifact_id, job_id, project_id, media_id, filename, content_type, size_bytes,
|
||||
created_at, expires_at, visibility, sha256, storage_status, source_skill_id,
|
||||
source_model_id, source_route_key, source_input_json, prompt_text, usage_json, ref_json
|
||||
FROM artifacts WHERE user_id = ? AND artifact_id = ?`,
|
||||
userID, artifactID,
|
||||
)
|
||||
return scanArtifact(row)
|
||||
}
|
||||
|
||||
func (r *sqliteRepository) ListArtifacts(userID, projectID, jobID string, limit, offset int) ([]artifact, int, error) {
|
||||
if strings.TrimSpace(userID) == "" {
|
||||
return nil, 0, errors.New("user_id is required")
|
||||
}
|
||||
where, args := buildArtifactFilters(userID, projectID, jobID)
|
||||
var total int
|
||||
if err := r.db.QueryRow(`SELECT COUNT(*) FROM artifacts`+where, args...).Scan(&total); err != nil {
|
||||
return nil, 0, fmt.Errorf("count artifacts: %w", err)
|
||||
}
|
||||
|
||||
argsWithPaging := append(append([]any{}, args...), limit, offset)
|
||||
rows, err := r.db.Query(
|
||||
`SELECT artifact_id, job_id, project_id, media_id, filename, content_type, size_bytes,
|
||||
created_at, expires_at, visibility, sha256, storage_status, source_skill_id,
|
||||
source_model_id, source_route_key, source_input_json, prompt_text, usage_json, ref_json
|
||||
FROM artifacts`+where+` ORDER BY datetime(created_at) DESC, artifact_id DESC LIMIT ? OFFSET ?`,
|
||||
argsWithPaging...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, 0, fmt.Errorf("list artifacts: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
items := make([]artifact, 0, limit)
|
||||
for rows.Next() {
|
||||
item, _, exists, err := scanArtifact(rows)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
if exists && item != nil {
|
||||
items = append(items, *item)
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, 0, fmt.Errorf("iterate artifacts: %w", err)
|
||||
}
|
||||
return items, total, nil
|
||||
}
|
||||
|
||||
func (r *sqliteRepository) findIdempotentJob(userID, idem string) (string, bool, error) {
|
||||
var jobID string
|
||||
err := r.db.QueryRow(
|
||||
@@ -543,6 +807,87 @@ func buildJobFilters(userID, status, skillID, projectID string) (string, []any)
|
||||
return " WHERE " + strings.Join(filters, " AND "), args
|
||||
}
|
||||
|
||||
func buildArtifactFilters(userID, projectID, jobID string) (string, []any) {
|
||||
filters := []string{"user_id = ?"}
|
||||
args := []any{userID}
|
||||
if projectID != "" {
|
||||
filters = append(filters, "project_id = ?")
|
||||
args = append(args, projectID)
|
||||
}
|
||||
if jobID != "" {
|
||||
filters = append(filters, "job_id = ?")
|
||||
args = append(args, jobID)
|
||||
}
|
||||
return " WHERE " + strings.Join(filters, " AND "), args
|
||||
}
|
||||
|
||||
func scanArtifact(row interface{ Scan(dest ...any) error }) (*artifact, resultRef, bool, error) {
|
||||
var (
|
||||
item artifact
|
||||
projectID, mediaID, expiresAt sql.NullString
|
||||
visibility, sha256, storageStatus sql.NullString
|
||||
sourceSkillID, sourceModelID, sourceRouteKey sql.NullString
|
||||
sourceInputJSON, promptText, usageJSON sql.NullString
|
||||
refJSON string
|
||||
)
|
||||
err := row.Scan(
|
||||
&item.ID,
|
||||
&item.JobID,
|
||||
&projectID,
|
||||
&mediaID,
|
||||
&item.Filename,
|
||||
&item.ContentType,
|
||||
&item.SizeBytes,
|
||||
&item.CreatedAt,
|
||||
&expiresAt,
|
||||
&visibility,
|
||||
&sha256,
|
||||
&storageStatus,
|
||||
&sourceSkillID,
|
||||
&sourceModelID,
|
||||
&sourceRouteKey,
|
||||
&sourceInputJSON,
|
||||
&promptText,
|
||||
&usageJSON,
|
||||
&refJSON,
|
||||
)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return nil, resultRef{}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, resultRef{}, false, fmt.Errorf("scan artifact: %w", err)
|
||||
}
|
||||
|
||||
item.ProjectID = projectID.String
|
||||
item.MediaID = mediaID.String
|
||||
item.ExpiresAt = expiresAt.String
|
||||
item.Visibility = visibility.String
|
||||
item.SHA256 = sha256.String
|
||||
item.StorageStatus = storageStatus.String
|
||||
item.SourceSkillID = sourceSkillID.String
|
||||
item.SourceModelID = sourceModelID.String
|
||||
item.SourceRouteKey = sourceRouteKey.String
|
||||
item.PromptText = promptText.String
|
||||
if sourceInputJSON.Valid {
|
||||
if err := unmarshalJSON(sourceInputJSON.String, &item.SourceInput); err != nil {
|
||||
return nil, resultRef{}, false, fmt.Errorf("decode artifact source input: %w", err)
|
||||
}
|
||||
}
|
||||
if usageJSON.Valid {
|
||||
if err := unmarshalJSON(usageJSON.String, &item.Usage); err != nil {
|
||||
return nil, resultRef{}, false, fmt.Errorf("decode artifact usage: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
var ref resultRef
|
||||
if err := unmarshalJSON(refJSON, &ref); err != nil {
|
||||
return nil, resultRef{}, false, fmt.Errorf("decode artifact ref: %w", err)
|
||||
}
|
||||
item.Ref = ref
|
||||
item.URL = strings.TrimSpace(ref.URL)
|
||||
return &item, ref, true, nil
|
||||
}
|
||||
|
||||
func scanJob(row interface{ Scan(dest ...any) error }) (*job, bool, error) {
|
||||
var (
|
||||
record job
|
||||
|
||||
Reference in New Issue
Block a user