From d0318292c157ccf1ff15a684a9dfaf1a4a6c36b5 Mon Sep 17 00:00:00 2001 From: wtgoku <2687861+wtgoku@users.noreply.github.com> Date: Sat, 18 Apr 2026 23:00:43 +0800 Subject: [PATCH] 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 --- docs/persistence.md | 96 +++++-- docs/stable-media-url-v1.md | 11 +- internal/server/artifact_list_test.go | 152 +++++++++++ internal/server/artifact_upload_test.go | 129 +++++++++ internal/server/media.go | 101 ++++++- internal/server/media_test.go | 98 +++++++ internal/server/repository.go | 11 + internal/server/server.go | 98 +++++-- internal/server/sqlite_repo.go | 345 ++++++++++++++++++++++++ internal/server/store.go | 174 ++++++++++-- internal/server/types.go | 33 ++- internal/server/video_test.go | 130 +++++++++ 12 files changed, 1303 insertions(+), 75 deletions(-) create mode 100644 internal/server/artifact_list_test.go diff --git a/docs/persistence.md b/docs/persistence.md index f940023..e92a6f3 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -14,32 +14,38 @@ 所以 `popiartServer` 不再重复持有： -- 本地 artifact metadata table +- provider 侧原始产物数据库 - 本地 job_logs table -它只保留三类本地状态： +它当前保留五类本地状态： - `sessions` - `jobs`（更准确说是 job refs） +- `artifacts` +- `media_records` - `skill_routes` -- `media blobs + metadata` +- `media blobs` ## 当前关系 ```text session -> jobs jobs -> result_refs_json +jobs -> artifacts +artifacts -> media_records project -> skill_routes -media -> local files + metadata json +media -> local files ``` 其中： - `session` 保存 PopiArt 登录态和对应的 `PopiNewAPI token` - `jobs` 保存 skill 语义、用户归属、上游引用和同步结果引用 -- `result_refs_json` 保存同步能力的结果引用；新结果会优先 re-host 到本地 media store +- `result_refs_json` 仍作为 job 级兼容引用保留 +- `artifacts` 保存用户可查询的 artifact 元数据与溯源快照 +- `media_records` 保存稳定 URL 所需的 media 元数据 - `skill_routes` 保存项目级路由覆盖 -- `media` 保存稳定 URL 所需的本地文件与元数据 +- `media blobs` 保存真实文件内容 ## 存储位置 @@ -62,16 +68,19 @@ POPIART_SQLITE_PATH=./data/popiart.db - `internal/server/repository.go` -当前只保留三层 repository： +当前 repository 分成五层： 1. `SessionRepository` 2. `JobRepository` 3. `RouteRepository` +4. `ArtifactRepository` +5. `MediaRepository` 本地开发版没有对象存储依赖，但现在有一个轻量 media 存储层： - blob 文件保存在 `POPIART_DATA_DIR/media/blobs/` -- metadata JSON 保存在 `POPIART_DATA_DIR/media/meta/` +- 兼容期 metadata JSON 保存在 `POPIART_DATA_DIR/media/meta/` +- SQLite 中的 `media_records` / `artifacts` 是当前主元数据存储 ## SQLite Schema @@ -115,6 +124,53 @@ CREATE TABLE jobs ( ); ``` +### media_records + +```sql +CREATE TABLE 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 +); +``` + +### artifacts + +```sql +CREATE TABLE 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 +); +``` + ### skill_routes ```sql @@ -148,31 +204,35 @@ CREATE TABLE skill_routes ( 1. 如果 `PopiNewAPI` 直接返回同步结果，例如 `b64_json` 或 `url` 2. `popiartServer` 会先把结果 re-host 到本地 media store 3. 再把本地 `local_path + media_id + stable url` 写进 `result_refs_json` -4. `GET /jobs/:id/artifacts` 时再从 `result_refs_json` 派生 artifact 列表 +4. 同时 upsert 到 `media_records` 和 `artifacts` +5. `GET /jobs/:id/artifacts` / `GET /v1/artifacts` 优先从 SQLite 读取 ### 拉取 artifact 1. server 从 `artifact_id` 反推出 `job_id + result index` -2. 读取 `jobs.result_refs_json` -3. 如果是本地 `local_path`，直接读取本地文件并流式返回 -4. 如果是旧的 `data_url`，直接解码并流式返回 -5. 如果是旧的远端 `url`，server 代理下载并返回 +2. 优先读取 `artifacts` 表中的 `ref_json` +3. 如果 SQLite 尚未命中，则从 `jobs.result_refs_json` 回填并重试 +4. 如果是本地 `local_path`，直接读取本地文件并流式返回 +5. 如果是旧的 `data_url`，直接解码并流式返回 +6. 如果是旧的远端 `url`，server 代理下载并返回 ### 读取 media 1. `POST /v1/media/upload` 会把本地文件写入 `media/blobs/` -2. server 同时写一份 metadata JSON 到 `media/meta/` -3. `GET /v1/media/:id` 返回 media 元数据 -4. `GET /v1/media/:id/content` 返回可供模型或客户端直接 fetch 的稳定内容路径 +2. server 同时写 SQLite `media_records`，并保留兼容 JSON meta +3. `GET /v1/media/:id` 优先读 SQLite，JSON 作为 fallback +4. `GET /v1/media/:id/content` 从 `local_path` 读取内容 ## 当前边界 - server 重启后： - `session` 会保留 - `job ref` 会保留 - - `artifact` 通过 `result_refs_json` 继续可读 + - `artifact` 通过 SQLite 主路径继续可读 + - 旧数据仍可通过 `result_refs_json` / JSON meta 懒回填 - 新 `media` 文件和 metadata 会继续可读 - 已存在的 `pending/running` job 仍然不会自动恢复执行 - 视频类 `upstream_task` 路径只预留了字段，后续再接 `PopiNewAPI` task 查询 - 旧数据里仍可能存在 `data_url` 或上游 `url` -- 新写入路径优先落本地 media store，从而给 artifact 补出稳定 `url` +- 旧 `media/meta/*.json` 仍保留作为兼容层，不再是主元数据来源 +- 新写入路径优先落 SQLite + 本地 media store，从而给 artifact 补出稳定 `url` diff --git a/docs/stable-media-url-v1.md b/docs/stable-media-url-v1.md index b5edc16..ec9a8c1 100644 --- a/docs/stable-media-url-v1.md +++ b/docs/stable-media-url-v1.md @@ -47,7 +47,14 @@ ### `GET /v1/media/:id/content` -读取稳定内容 URL。这个接口默认允许匿名 GET，以便模型提供商可以直接 fetch。 +读取稳定内容 URL。 + +当前语义改为： + +- 元数据接口仍然要求登录态，并要求 media 属于当前用户 +- 内容 URL 使用短期签名 query 参数作为 capability URL +- 外部模型可以直接 fetch 已签名的 URL，不需要 bearer session +- 裸的 `/content` 路径默认不可直接匿名读取 ## Artifact 行为变化 @@ -120,7 +127,7 @@ V1 的本地开发版不引入 S3/R2/OSS，而是用本地文件系统： ## 后续阶段 -后续如果要走生产化路线，可以保持接口不变，只把存储实现替换成对象存储： +后续如果要走生产化路线，可以保持接口不变，只把签名 URL 和存储实现替换成对象存储： - 本地 `media/blobs/` -> S3 / R2 / OSS / COS - metadata JSON -> 独立 media table 或对象存储元数据 diff --git a/internal/server/artifact_list_test.go b/internal/server/artifact_list_test.go new file mode 100644 index 0000000..9fd4bea --- /dev/null +++ b/internal/server/artifact_list_test.go @@ -0,0 +1,152 @@ +package server + +import ( + "bytes" + "encoding/json" + "mime/multipart" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" +) + +func TestArtifactsListReturnsOnlyOwnerArtifacts(t *testing.T) { + cfg := Config{ + SQLitePath: filepath.Join(t.TempDir(), "popiart.db"), + SkillhubDir: makeEmptySkillhub(t), + SessionSecret: "test-secret", + } + server, err := NewWithConfig(cfg) + if err != nil { + t.Fatalf("NewWithConfig: %v", err) + } + srv := httptest.NewServer(server.Handler()) + defer srv.Close() + server.cfg.PublicBaseURL = srv.URL + + ownerToken, _, ok, err := server.store.createSession("sk-artifacts-owner") + if err != nil { + t.Fatalf("createSession owner: %v", err) + } + if !ok { + t.Fatal("expected owner session creation to succeed") + } + otherToken, _, ok, err := server.store.createSession("sk-artifacts-other") + if err != nil { + t.Fatalf("createSession other: %v", err) + } + if !ok { + t.Fatal("expected other session creation to succeed") + } + + ownerArtifact := uploadArtifactForListTest(t, server, ownerToken, "proj_owner", "owner.png") + otherArtifact := uploadArtifactForListTest(t, server, otherToken, "proj_other", "other.png") + + req := httptest.NewRequest(http.MethodGet, "/v1/artifacts", nil) + req.Header.Set("Authorization", "Bearer "+ownerToken) + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("expected list 200, got %d body=%s", rec.Code, rec.Body.String()) + } + + var envelope struct { + OK bool `json:"ok"` + Data struct { + Items []artifact `json:"items"` + Total int `json:"total"` + Limit int `json:"limit"` + Offset int `json:"offset"` + } `json:"data"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &envelope); err != nil { + t.Fatalf("decode artifacts list: %v body=%s", err, rec.Body.String()) + } + if !envelope.OK { + t.Fatalf("expected ok artifacts list, got %s", rec.Body.String()) + } + if envelope.Data.Total != 1 { + t.Fatalf("expected total 1, got %d payload=%s", envelope.Data.Total, rec.Body.String()) + } + if len(envelope.Data.Items) != 1 { + t.Fatalf("expected one artifact, got %d payload=%s", len(envelope.Data.Items), rec.Body.String()) + } + if envelope.Data.Items[0].ID != ownerArtifact.ID { + t.Fatalf("expected owner artifact %q, got %#v", ownerArtifact.ID, envelope.Data.Items[0]) + } + if envelope.Data.Items[0].ID == otherArtifact.ID { + t.Fatalf("expected foreign artifact to be filtered out, got %#v", envelope.Data.Items[0]) + } + + filterReq := httptest.NewRequest(http.MethodGet, "/v1/artifacts?project_id=proj_owner", nil) + filterReq.Header.Set("Authorization", "Bearer "+ownerToken) + filterRec := httptest.NewRecorder() + server.Handler().ServeHTTP(filterRec, filterReq) + if filterRec.Code != http.StatusOK { + t.Fatalf("expected filtered list 200, got %d body=%s", filterRec.Code, filterRec.Body.String()) + } + if err := json.Unmarshal(filterRec.Body.Bytes(), &envelope); err != nil { + t.Fatalf("decode filtered artifacts list: %v body=%s", err, filterRec.Body.String()) + } + if envelope.Data.Total != 1 || len(envelope.Data.Items) != 1 { + t.Fatalf("expected one filtered owner artifact, got payload=%s", filterRec.Body.String()) + } + + foreignReq := httptest.NewRequest(http.MethodGet, "/v1/artifacts?job_id="+otherArtifact.JobID, nil) + foreignReq.Header.Set("Authorization", "Bearer "+ownerToken) + foreignRec := httptest.NewRecorder() + server.Handler().ServeHTTP(foreignRec, foreignReq) + if foreignRec.Code != http.StatusOK { + t.Fatalf("expected foreign job list 200, got %d body=%s", foreignRec.Code, foreignRec.Body.String()) + } + if err := json.Unmarshal(foreignRec.Body.Bytes(), &envelope); err != nil { + t.Fatalf("decode foreign job artifacts list: %v body=%s", err, foreignRec.Body.String()) + } + if envelope.Data.Total != 0 || len(envelope.Data.Items) != 0 { + t.Fatalf("expected foreign job artifacts to be hidden, got payload=%s", foreignRec.Body.String()) + } +} + +func uploadArtifactForListTest(t *testing.T, server *Server, sessionToken, projectID, filename string) artifact { + t.Helper() + + imageBytes := tinyPNG(t) + var body bytes.Buffer + writer := multipart.NewWriter(&body) + if projectID != "" { + if err := writer.WriteField("project_id", projectID); err != nil { + t.Fatalf("write project_id: %v", err) + } + } + part, err := writer.CreateFormFile("file", filename) + if err != nil { + t.Fatalf("create form file: %v", err) + } + if _, err := part.Write(imageBytes); err != nil { + t.Fatalf("write image bytes: %v", err) + } + if err := writer.Close(); err != nil { + t.Fatalf("close multipart writer: %v", err) + } + + req := httptest.NewRequest(http.MethodPost, "/v1/artifacts/upload", &body) + req.Header.Set("Authorization", "Bearer "+sessionToken) + req.Header.Set("Content-Type", writer.FormDataContentType()) + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusCreated { + t.Fatalf("expected artifact upload 201, got %d body=%s", rec.Code, rec.Body.String()) + } + + var envelope struct { + OK bool `json:"ok"` + Data artifact `json:"data"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &envelope); err != nil { + t.Fatalf("decode artifact upload: %v body=%s", err, rec.Body.String()) + } + if !envelope.OK { + t.Fatalf("expected ok artifact upload, got %s", rec.Body.String()) + } + return envelope.Data +} diff --git a/internal/server/artifact_upload_test.go b/internal/server/artifact_upload_test.go index 3c05168..bede99d 100644 --- a/internal/server/artifact_upload_test.go +++ b/internal/server/artifact_upload_test.go @@ -9,6 +9,7 @@ import ( "net/http" "net/http/httptest" "path/filepath" + "strings" "testing" ) @@ -96,6 +97,15 @@ func TestArtifactUploadCreatesReadableArtifactForSourceArtifactID(t *testing.T) if envelope.Data.StorageStatus != "ready" { t.Fatalf("expected storage status ready, got %#v", envelope.Data.StorageStatus) } + if envelope.Data.SourceSkillID != "popiskill-artifact-upload-local-v1" { + t.Fatalf("expected source skill id, got %#v", envelope.Data.SourceSkillID) + } + if envelope.Data.SourceRouteKey != "artifact.upload" { + t.Fatalf("expected source route key, got %#v", envelope.Data.SourceRouteKey) + } + if envelope.Data.SourceInput["role"] != "source" { + t.Fatalf("expected source input role, got %#v", envelope.Data.SourceInput) + } contentResp, err := http.Get(envelope.Data.URL) if err != nil { @@ -115,6 +125,7 @@ func TestArtifactUploadCreatesReadableArtifactForSourceArtifactID(t *testing.T) editRef, err := server.resolveImageToImageReference(context.Background(), &job{ UserID: current.User.ID, + SessionID: current.Token, UpstreamKey: current.UpstreamKey, SkillID: "popiskill-image-img2img-basic-v1", }, map[string]any{ @@ -129,6 +140,39 @@ func TestArtifactUploadCreatesReadableArtifactForSourceArtifactID(t *testing.T) if !bytes.Equal(editRef.Content, imageBytes) { t.Fatal("expected resolved reference content to match uploaded bytes") } + + unsignedResp, err := http.Get(strings.Split(envelope.Data.URL, "?")[0]) + if err != nil { + t.Fatalf("GET unsigned artifact media content: %v", err) + } + defer unsignedResp.Body.Close() + if unsignedResp.StatusCode != http.StatusUnauthorized { + t.Fatalf("expected unsigned artifact media content 401, got %d", unsignedResp.StatusCode) + } + + repo, ok := server.store.artifacts.(*sqliteRepository) + if !ok { + t.Fatalf("expected sqliteRepository, got %T", server.store.artifacts) + } + if _, err := repo.db.Exec(`UPDATE jobs SET result_refs_json = NULL WHERE job_id = ?`, envelope.Data.JobID); err != nil { + t.Fatalf("clear job result refs: %v", err) + } + + getReq := httptest.NewRequest(http.MethodGet, "/v1/artifacts/"+envelope.Data.ID, nil) + getReq.Header.Set("Authorization", "Bearer "+sessionToken) + getRec := httptest.NewRecorder() + server.Handler().ServeHTTP(getRec, getReq) + if getRec.Code != http.StatusOK { + t.Fatalf("expected sqlite-backed artifact get 200, got %d body=%s", getRec.Code, getRec.Body.String()) + } + + listReq := httptest.NewRequest(http.MethodGet, "/v1/artifacts?job_id="+envelope.Data.JobID, nil) + listReq.Header.Set("Authorization", "Bearer "+sessionToken) + listRec := httptest.NewRecorder() + server.Handler().ServeHTTP(listRec, listReq) + if listRec.Code != http.StatusOK { + t.Fatalf("expected sqlite-backed artifacts list 200, got %d body=%s", listRec.Code, listRec.Body.String()) + } } func TestResolveImageToImageReferenceAcceptsCanonicalImageURL(t *testing.T) { @@ -169,3 +213,88 @@ func TestResolveImageToImageReferenceAcceptsCanonicalImageURL(t *testing.T) { t.Fatal("expected resolved canonical image content to match downloaded bytes") } } + +func TestResolveImageToImageReferenceAcceptsSignedSameOriginMediaURL(t *testing.T) { + cfg := Config{ + SQLitePath: filepath.Join(t.TempDir(), "popiart.db"), + SkillhubDir: makeEmptySkillhub(t), + SessionSecret: "test-secret", + } + server, err := NewWithConfig(cfg) + if err != nil { + t.Fatalf("NewWithConfig: %v", err) + } + srv := httptest.NewServer(server.Handler()) + defer srv.Close() + server.cfg.PublicBaseURL = srv.URL + + sessionToken, _, ok, err := server.store.createSession("sk-same-origin-user") + if err != nil { + t.Fatalf("createSession: %v", err) + } + if !ok { + t.Fatal("expected session creation to succeed") + } + current, exists, err := server.store.session(sessionToken) + if err != nil { + t.Fatalf("load session: %v", err) + } + if !exists { + t.Fatal("expected stored session") + } + + imageBytes := tinyPNG(t) + var body bytes.Buffer + writer := multipart.NewWriter(&body) + part, err := writer.CreateFormFile("file", "same-origin.png") + if err != nil { + t.Fatalf("create form file: %v", err) + } + if _, err := part.Write(imageBytes); err != nil { + t.Fatalf("write form file: %v", err) + } + if err := writer.Close(); err != nil { + t.Fatalf("close multipart writer: %v", err) + } + + req := httptest.NewRequest(http.MethodPost, "/v1/media/upload", &body) + req.Header.Set("Authorization", "Bearer "+sessionToken) + req.Header.Set("Content-Type", writer.FormDataContentType()) + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusCreated { + t.Fatalf("expected media upload 201, got %d body=%s", rec.Code, rec.Body.String()) + } + + var envelope struct { + OK bool `json:"ok"` + Data media `json:"data"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &envelope); err != nil { + t.Fatalf("decode media upload response: %v body=%s", err, rec.Body.String()) + } + if !envelope.OK || envelope.Data.URL == "" { + t.Fatalf("expected signed media url, got %s", rec.Body.String()) + } + + editRef, err := server.resolveImageToImageReference(context.Background(), &job{ + UserID: current.User.ID, + SessionID: current.Token, + UpstreamKey: current.UpstreamKey, + SkillID: "popiskill-image-img2img-basic-v1", + }, map[string]any{ + "image": envelope.Data.URL, + }) + if err != nil { + t.Fatalf("resolveImageToImageReference same-origin: %v", err) + } + if editRef.URL != envelope.Data.URL { + t.Fatalf("expected resolved URL to round-trip, got %q", editRef.URL) + } + if editRef.ContentType != "image/png" { + t.Fatalf("expected image/png, got %q", editRef.ContentType) + } + if !bytes.Equal(editRef.Content, imageBytes) { + t.Fatal("expected same-origin signed media bytes to match uploaded file") + } +} diff --git a/internal/server/media.go b/internal/server/media.go index fcbc801..726412e 100644 --- a/internal/server/media.go +++ b/internal/server/media.go @@ -2,7 +2,9 @@ package server import ( "context" + "crypto/hmac" "crypto/sha256" + "encoding/base64" "encoding/hex" "encoding/json" "fmt" @@ -10,6 +12,7 @@ import ( "net/url" "os" "path/filepath" + "strconv" "strings" "time" ) @@ -59,11 +62,95 @@ func publicBaseURL(cfg Config) string { return strings.TrimRight(base, "/") } +func mediaContentPath(mediaID string) string { + return "/v1/media/" + url.PathEscape(strings.TrimSpace(mediaID)) + "/content" +} + func mediaContentURL(cfg Config, mediaID string) string { + return publicBaseURL(cfg) + mediaContentPath(mediaID) +} + +func mediaURLTTL() time.Duration { + return 24 * time.Hour +} + +func mediaURLExpiry(now time.Time) time.Time { + if now.IsZero() { + now = time.Now().UTC() + } + return now.UTC().Add(mediaURLTTL()) +} + +func mediaURLSecret(cfg Config) []byte { + return []byte(defaultString(strings.TrimSpace(cfg.SessionSecret), "popiart-dev-session-secret")) +} + +func signedMediaContentURL(cfg Config, mediaID string, now time.Time) string { + expiresAt := mediaURLExpiry(now) + baseURL := mediaContentURL(cfg, mediaID) + exp := strconv.FormatInt(expiresAt.Unix(), 10) + sig := signMediaAccess(cfg, mediaID, exp) + return baseURL + "?exp=" + url.QueryEscape(exp) + "&sig=" + url.QueryEscape(sig) +} + +func signMediaAccess(cfg Config, mediaID, exp string) string { + mac := hmac.New(sha256.New, mediaURLSecret(cfg)) + _, _ = mac.Write([]byte(strings.TrimSpace(mediaID))) + _, _ = mac.Write([]byte("\n")) + _, _ = mac.Write([]byte(strings.TrimSpace(exp))) + return base64.RawURLEncoding.EncodeToString(mac.Sum(nil)) +} + +func validateSignedMediaAccess(cfg Config, mediaID, exp, sig string, now time.Time) bool { + exp = strings.TrimSpace(exp) + sig = strings.TrimSpace(sig) + if exp == "" || sig == "" { + return false + } + unix, err := strconv.ParseInt(exp, 10, 64) + if err != nil { + return false + } + if now.IsZero() { + now = time.Now().UTC() + } + if now.UTC().After(time.Unix(unix, 0).UTC()) { + return false + } + expected := signMediaAccess(cfg, mediaID, exp) + return hmac.Equal([]byte(expected), []byte(sig)) +} + +func mediaView(cfg Config, record mediaRecord, now time.Time) media { + view := record.media + view.URL = signedMediaContentURL(cfg, record.ID, now) + return view +} + +func artifactView(cfg Config, item artifact, now time.Time) artifact { + view := item + if strings.TrimSpace(view.MediaID) != "" { + view.URL = signedMediaContentURL(cfg, view.MediaID, now) + } + return view +} + +func artifactViews(cfg Config, items []artifact, now time.Time) []artifact { + if len(items) == 0 { + return nil + } + out := make([]artifact, 0, len(items)) + for _, item := range items { + out = append(out, artifactView(cfg, item, now)) + } + return out +} + +func mediaContentURLCanonical(cfg Config, mediaID string) string { return publicBaseURL(cfg) + "/v1/media/" + url.PathEscape(strings.TrimSpace(mediaID)) + "/content" } -func persistMediaContent(cfg Config, userID, projectID, artifactID, filename, contentType, visibility string, content []byte, mediaID string) (mediaRecord, error) { +func persistMediaContent(cfg Config, repo MediaRepository, userID, projectID, artifactID, filename, contentType, visibility string, content []byte, mediaID string) (mediaRecord, error) { if len(content) == 0 { return mediaRecord{}, fmt.Errorf("media content is required") } @@ -105,7 +192,7 @@ func persistMediaContent(cfg Config, userID, projectID, artifactID, filename, co ContentType: contentType, SizeBytes: int64(len(content)), CreatedAt: time.Now().UTC().Format(time.RFC3339), - URL: mediaContentURL(cfg, mediaID), + URL: mediaContentURLCanonical(cfg, mediaID), Visibility: visibility, SHA256: hex.EncodeToString(sum[:]), }, @@ -120,10 +207,15 @@ func persistMediaContent(cfg Config, userID, projectID, artifactID, filename, co if err := os.WriteFile(mediaMetaPath(cfg, mediaID), metaBytes, 0o644); err != nil { return mediaRecord{}, err } + if repo != nil { + if err := repo.UpsertMedia(record); err != nil { + return mediaRecord{}, err + } + } return record, nil } -func loadMediaRecord(cfg Config, mediaID string) (*mediaRecord, bool, error) { +func loadMediaRecordFromJSON(cfg Config, mediaID string) (*mediaRecord, bool, error) { mediaID = strings.TrimSpace(mediaID) if mediaID == "" { return nil, false, nil @@ -162,8 +254,7 @@ func (s *Server) persistResultRefs(ctx context.Context, record *job, refs []resu filename = inferFilenameFromRef(ref, buildArtifactID(record.JobID, idx)) } artifactID := buildArtifactID(record.JobID, idx) - mediaRecord, err := persistMediaContent( - s.cfg, + mediaRecord, err := s.store.persistMediaContent( record.UserID, record.ProjectID, artifactID, diff --git a/internal/server/media_test.go b/internal/server/media_test.go index 7a2a48c..afec782 100644 --- a/internal/server/media_test.go +++ b/internal/server/media_test.go @@ -7,7 +7,9 @@ import ( "mime/multipart" "net/http" "net/http/httptest" + "os" "path/filepath" + "strings" "testing" ) @@ -87,6 +89,9 @@ func TestMediaUploadGetAndContent(t *testing.T) { if getRec.Code != http.StatusOK { t.Fatalf("expected media get 200, got %d body=%s", getRec.Code, getRec.Body.String()) } + if strings.TrimSpace(envelope.Data.URL) == "" { + t.Fatalf("expected signed media url, got %#v", envelope.Data) + } contentResp, err := http.Get(envelope.Data.URL) if err != nil { @@ -103,4 +108,97 @@ func TestMediaUploadGetAndContent(t *testing.T) { if !bytes.Equal(content, imageBytes) { t.Fatal("expected media content bytes to match uploaded file") } + + unsignedResp, err := http.Get(strings.Split(envelope.Data.URL, "?")[0]) + if err != nil { + t.Fatalf("GET unsigned media content: %v", err) + } + defer unsignedResp.Body.Close() + if unsignedResp.StatusCode != http.StatusUnauthorized { + t.Fatalf("expected unsigned media content 401, got %d", unsignedResp.StatusCode) + } + + otherSession, _, ok, err := server.store.createSession("sk-media-user-other") + if err != nil { + t.Fatalf("createSession other: %v", err) + } + if !ok { + t.Fatal("expected second session creation to succeed") + } + otherReq := httptest.NewRequest(http.MethodGet, "/v1/media/"+envelope.Data.ID, nil) + otherReq.Header.Set("Authorization", "Bearer "+otherSession) + otherRec := httptest.NewRecorder() + server.Handler().ServeHTTP(otherRec, otherReq) + if otherRec.Code != http.StatusNotFound { + t.Fatalf("expected foreign media metadata 404, got %d body=%s", otherRec.Code, otherRec.Body.String()) + } +} + +func TestMediaGetFallsBackToSQLiteWhenJSONMetaIsMissing(t *testing.T) { + cfg := Config{ + SQLitePath: filepath.Join(t.TempDir(), "popiart.db"), + SkillhubDir: makeEmptySkillhub(t), + SessionSecret: "test-secret", + } + server, err := NewWithConfig(cfg) + if err != nil { + t.Fatalf("NewWithConfig: %v", err) + } + srv := httptest.NewServer(server.Handler()) + defer srv.Close() + server.cfg.PublicBaseURL = srv.URL + + sessionToken, _, ok, err := server.store.createSession("sk-media-sqlite") + if err != nil { + t.Fatalf("createSession: %v", err) + } + if !ok { + t.Fatal("expected session creation to succeed") + } + + imageBytes := tinyPNG(t) + var body bytes.Buffer + writer := multipart.NewWriter(&body) + part, err := writer.CreateFormFile("file", "sqlite-only.png") + if err != nil { + t.Fatalf("create form file: %v", err) + } + if _, err := part.Write(imageBytes); err != nil { + t.Fatalf("write form file: %v", err) + } + if err := writer.Close(); err != nil { + t.Fatalf("close multipart writer: %v", err) + } + + req := httptest.NewRequest(http.MethodPost, "/v1/media/upload", &body) + req.Header.Set("Authorization", "Bearer "+sessionToken) + req.Header.Set("Content-Type", writer.FormDataContentType()) + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusCreated { + t.Fatalf("expected media upload 201, got %d body=%s", rec.Code, rec.Body.String()) + } + + var envelope struct { + OK bool `json:"ok"` + Data media `json:"data"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &envelope); err != nil { + t.Fatalf("decode media upload response: %v body=%s", err, rec.Body.String()) + } + if !envelope.OK || envelope.Data.ID == "" { + t.Fatalf("expected media upload payload, got %s", rec.Body.String()) + } + + if err := os.Remove(mediaMetaPath(server.cfg, envelope.Data.ID)); err != nil { + t.Fatalf("remove media json meta: %v", err) + } + + getReq := httptest.NewRequest(http.MethodGet, "/v1/media/"+envelope.Data.ID, nil) + getReq.Header.Set("Authorization", "Bearer "+sessionToken) + getRec := httptest.NewRecorder() + server.Handler().ServeHTTP(getRec, getReq) + if getRec.Code != http.StatusOK { + t.Fatalf("expected media get 200 from sqlite, got %d body=%s", getRec.Code, getRec.Body.String()) + } } diff --git a/internal/server/repository.go b/internal/server/repository.go index fa52032..5b84a93 100644 --- a/internal/server/repository.go +++ b/internal/server/repository.go @@ -25,3 +25,14 @@ type RouteRepository interface { SetRoute(projectID, routeKey, modelID string) error UnsetRoute(projectID, routeKey string) error } + +type MediaRepository interface { + UpsertMedia(record mediaRecord) error + GetMedia(mediaID string) (*mediaRecord, bool, error) +} + +type ArtifactRepository interface { + UpsertArtifacts(userID, projectID string, items []artifact) error + GetArtifact(userID, artifactID string) (*artifact, resultRef, bool, error) + ListArtifacts(userID, projectID, jobID string, limit, offset int) ([]artifact, int, error) +} diff --git a/internal/server/server.go b/internal/server/server.go index 508cc96..a8462a7 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -63,6 +63,7 @@ func (s *Server) routes() { s.mux.HandleFunc("/v1/jobs/", s.handleJob) s.mux.HandleFunc("/v1/media/upload", s.handleMediaUpload) s.mux.HandleFunc("/v1/media/", s.handleMedia) + s.mux.HandleFunc("/v1/artifacts", s.handleArtifacts) s.mux.HandleFunc("/v1/artifacts/upload", s.handleArtifactUpload) s.mux.HandleFunc("/v1/artifacts/", s.handleArtifact) s.mux.HandleFunc("/v1/budget", s.handleBudget) @@ -409,6 +410,34 @@ func (s *Server) handleJob(w http.ResponseWriter, r *http.Request) { notFound(w) } +func (s *Server) handleArtifacts(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + methodNotAllowed(w) + return + } + current, ok := s.authenticateSession(w, r) + if !ok { + return + } + + projectID := strings.TrimSpace(r.URL.Query().Get("project_id")) + jobID := strings.TrimSpace(r.URL.Query().Get("job_id")) + limit := intQuery(r, "limit", 20) + offset := intQuery(r, "offset", 0) + + items, total, err := s.store.listArtifacts(current.User.ID, projectID, jobID, limit, offset) + if err != nil { + writeInternalError(w, "failed to list artifacts", err) + return + } + writeData(w, http.StatusOK, map[string]any{ + "items": artifactViews(s.cfg, items, time.Now().UTC()), + "total": total, + "limit": limit, + "offset": offset, + }) +} + func (s *Server) handleArtifactUpload(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { methodNotAllowed(w) @@ -504,8 +533,7 @@ func (s *Server) handleArtifactUpload(w http.ResponseWriter, r *http.Request) { return } - refMedia, err := persistMediaContent( - s.cfg, + refMedia, err := s.store.persistMediaContent( current.User.ID, projectID, buildArtifactID(record.JobID, 0), @@ -546,7 +574,7 @@ func (s *Server) handleArtifactUpload(w http.ResponseWriter, r *http.Request) { writeInternalError(w, "uploaded artifact missing after persistence", errors.New("artifact not found")) return } - writeData(w, http.StatusCreated, item) + writeData(w, http.StatusCreated, artifactView(s.cfg, *item, time.Now().UTC())) } func (s *Server) handleMediaUpload(w http.ResponseWriter, r *http.Request) { @@ -612,8 +640,7 @@ func (s *Server) handleMediaUpload(w http.ResponseWriter, r *http.Request) { } filename = sanitizeFilename(filename) - record, err := persistMediaContent( - s.cfg, + record, err := s.store.persistMediaContent( current.User.ID, strings.TrimSpace(r.FormValue("project_id")), "", @@ -627,7 +654,7 @@ func (s *Server) handleMediaUpload(w http.ResponseWriter, r *http.Request) { writeInternalError(w, "failed to persist media", err) return } - writeData(w, http.StatusCreated, record.media) + writeData(w, http.StatusCreated, mediaView(s.cfg, record, time.Now().UTC())) } func (s *Server) handleMedia(w http.ResponseWriter, r *http.Request) { @@ -638,7 +665,7 @@ func (s *Server) handleMedia(w http.ResponseWriter, r *http.Request) { return } - record, exists, err := loadMediaRecord(s.cfg, parts[0]) + record, exists, err := s.store.getMedia(parts[0]) if err != nil { writeInternalError(w, "failed to load media", err) return @@ -649,6 +676,22 @@ func (s *Server) handleMedia(w http.ResponseWriter, r *http.Request) { } if len(parts) == 2 && parts[1] == "content" && r.Method == http.MethodGet { + if !validateSignedMediaAccess(s.cfg, record.ID, r.URL.Query().Get("exp"), r.URL.Query().Get("sig"), time.Now().UTC()) { + current, ok, err := s.sessionFromRequest(r) + if err != nil { + writeInternalError(w, "failed to load session", err) + return + } + if !ok { + writeError(w, http.StatusUnauthorized, "UNAUTHENTICATED", "missing bearer key or signed media url", nil) + return + } + if current.User.ID != record.UserID { + notFound(w) + return + } + } + file, err := os.Open(record.LocalPath) if err != nil { writeInternalError(w, "failed to read media content", err) @@ -676,7 +719,7 @@ func (s *Server) handleMedia(w http.ResponseWriter, r *http.Request) { return } if len(parts) == 1 && r.Method == http.MethodGet { - writeData(w, http.StatusOK, record.media) + writeData(w, http.StatusOK, mediaView(s.cfg, *record, time.Now().UTC())) return } @@ -707,7 +750,7 @@ func (s *Server) handleArtifact(w http.ResponseWriter, r *http.Request) { } if len(parts) == 1 && r.Method == http.MethodGet { - writeData(w, http.StatusOK, item) + writeData(w, http.StatusOK, artifactView(s.cfg, *item, time.Now().UTC())) return } @@ -1454,7 +1497,7 @@ func (s *Server) resolveImageToImageReferences(ctx context.Context, record *job, if refURL == "" { return nil, fmt.Errorf("reference image is required") } - ref, err := s.downloadReferenceImage(ctx, refURL) + ref, err := s.downloadReferenceImage(ctx, record.SessionID, refURL) if err != nil { return nil, err } @@ -1503,7 +1546,7 @@ func (s *Server) resolveVideoReferences(ctx context.Context, record *job, input refs := make([]imageEditReference, 0, len(urls)) for _, rawURL := range urls { - ref, err := s.downloadReferenceImage(ctx, rawURL) + ref, err := s.downloadReferenceImage(ctx, record.SessionID, rawURL) if err != nil { return nil, err } @@ -1565,11 +1608,14 @@ func (s *Server) waitForVideoTask(ctx context.Context, token, taskID string) (*v } } -func (s *Server) downloadReferenceImage(ctx context.Context, rawURL string) (imageEditReference, error) { +func (s *Server) downloadReferenceImage(ctx context.Context, sessionToken, rawURL string) (imageEditReference, error) { req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil) if err != nil { return imageEditReference{}, err } + if sessionToken = strings.TrimSpace(sessionToken); sessionToken != "" && shouldAttachAuthHeader(publicBaseURL(s.cfg), rawURL) { + req.Header.Set("Authorization", "Bearer "+sessionToken) + } resp, err := s.newapi.httpClient.Do(req) if err != nil { return imageEditReference{}, err @@ -1709,7 +1755,7 @@ func (s *Server) resolveArtifactImageReference(ctx context.Context, record *job, Filename: item.Filename, ContentType: defaultString(contentType, item.ContentType), Content: content, - URL: item.URL, + URL: artifactView(s.cfg, *item, time.Now().UTC()).URL, }, nil } @@ -1833,23 +1879,33 @@ func buildVideoResultRef(baseURL, taskID, modelID string, task *videoTaskResult) const defaultAliceReferenceURL = "http://8.136.121.101:8790/media/Character_id_card/alice.jpg" func (s *Server) authenticateSession(w http.ResponseWriter, r *http.Request) (session, bool) { - token, ok := bearerToken(r) - if !ok { - writeError(w, http.StatusUnauthorized, "UNAUTHENTICATED", "missing bearer key", nil) - return session{}, false - } - current, exists, err := s.store.session(token) + current, ok, err := s.sessionFromRequest(r) if err != nil { writeInternalError(w, "failed to load session", err) return session{}, false } - if !exists { - writeError(w, http.StatusUnauthorized, "UNAUTHENTICATED", "key invalid or expired", nil) + if !ok { + writeError(w, http.StatusUnauthorized, "UNAUTHENTICATED", "missing bearer key", nil) return session{}, false } return current, true } +func (s *Server) sessionFromRequest(r *http.Request) (session, bool, error) { + token, ok := bearerToken(r) + if !ok { + return session{}, false, nil + } + current, exists, err := s.store.session(token) + if err != nil { + return session{}, false, err + } + if !exists { + return session{}, false, nil + } + return current, true, nil +} + func bearerToken(r *http.Request) (string, bool) { header := strings.TrimSpace(r.Header.Get("Authorization")) if header == "" || !strings.HasPrefix(header, "Bearer ") { diff --git a/internal/server/sqlite_repo.go b/internal/server/sqlite_repo.go index f3ca266..a50c2bd 100644 --- a/internal/server/sqlite_repo.go +++ b/internal/server/sqlite_repo.go @@ -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 diff --git a/internal/server/store.go b/internal/server/store.go index 23e0837..75af74e 100644 --- a/internal/server/store.go +++ b/internal/server/store.go @@ -13,9 +13,12 @@ import ( ) type store struct { + cfg Config sessions SessionRepository jobs JobRepository routes RouteRepository + media MediaRepository + artifacts ArtifactRepository skills []skill projects []project sessionTTL time.Duration @@ -36,9 +39,12 @@ func newStore(cfg Config) (*store, error) { } return &store{ + cfg: cfg, sessions: repo, jobs: repo, routes: repo, + media: repo, + artifacts: repo, skills: skills, projects: projects, sessionTTL: 30 * 24 * time.Hour, @@ -245,7 +251,14 @@ func (s *store) failJob(jobID, code, message string, details map[string]any) err } func (s *store) completeJobWithResults(jobID string, refs []resultRef, usage map[string]any) error { - return s.jobs.CompleteSyncResult(jobID, refs, cloneMap(usage)) + if err := s.jobs.CompleteSyncResult(jobID, refs, cloneMap(usage)); err != nil { + return err + } + record, exists, err := s.jobs.GetJob(jobID) + if err != nil || !exists || record == nil { + return err + } + return s.syncArtifactsForJob(record) } func (s *store) artifactsForJob(userID, jobID string) ([]artifact, bool, error) { @@ -253,10 +266,28 @@ func (s *store) artifactsForJob(userID, jobID string) ([]artifact, bool, error) if err != nil || !exists { return nil, exists, err } + if err := s.syncArtifactsForJob(record); err != nil { + return nil, true, err + } + if s.artifacts != nil { + items, _, err := s.artifacts.ListArtifacts(userID, "", jobID, 1000, 0) + if err != nil { + return nil, true, err + } + if len(items) > 0 || len(record.ResultRefs) == 0 { + return items, true, nil + } + } return buildArtifacts(record), true, nil } func (s *store) artifactRef(userID, artifactID string) (*artifact, resultRef, bool, error) { + if s.artifacts != nil { + item, ref, exists, err := s.artifacts.GetArtifact(userID, artifactID) + if err != nil || exists { + return item, ref, exists, err + } + } jobID, idx, ok := parseArtifactID(artifactID) if !ok { return nil, resultRef{}, false, nil @@ -265,6 +296,15 @@ func (s *store) artifactRef(userID, artifactID string) (*artifact, resultRef, bo if err != nil || !exists { return nil, resultRef{}, exists, err } + if err := s.syncArtifactsForJob(record); err != nil { + return nil, resultRef{}, true, err + } + if s.artifacts != nil { + item, ref, exists, err := s.artifacts.GetArtifact(userID, artifactID) + if err != nil || exists { + return item, ref, exists, err + } + } items := buildArtifacts(record) if idx < 0 || idx >= len(items) || idx >= len(record.ResultRefs) { return nil, resultRef{}, false, nil @@ -273,6 +313,23 @@ func (s *store) artifactRef(userID, artifactID string) (*artifact, resultRef, bo return &item, record.ResultRefs[idx], true, nil } +func (s *store) listArtifacts(userID, projectID, jobID string, limit, offset int) ([]artifact, int, error) { + if limit <= 0 { + limit = 20 + } + if offset < 0 { + offset = 0 + } + + if err := s.backfillArtifacts(userID, strings.TrimSpace(projectID), strings.TrimSpace(jobID)); err != nil { + return nil, 0, err + } + if s.artifacts != nil { + return s.artifacts.ListArtifacts(userID, projectID, jobID, limit, offset) + } + return nil, 0, nil +} + func (s *store) routesForProject(projectID string) (map[string]string, error) { return s.routes.GetRoutes(projectID) } @@ -285,6 +342,71 @@ func (s *store) unsetRouteOverride(projectID, routeKey string) error { return s.routes.UnsetRoute(projectID, normalizeRouteKey(routeKey)) } +func (s *store) persistMediaContent(userID, projectID, artifactID, filename, contentType, visibility string, content []byte, mediaID string) (mediaRecord, error) { + return persistMediaContent(s.cfg, s.media, userID, projectID, artifactID, filename, contentType, visibility, content, mediaID) +} + +func (s *store) getMedia(mediaID string) (*mediaRecord, bool, error) { + if s.media != nil { + record, exists, err := s.media.GetMedia(mediaID) + if err != nil || exists { + return record, exists, err + } + } + record, exists, err := loadMediaRecordFromJSON(s.cfg, mediaID) + if err != nil || !exists || record == nil { + return record, exists, err + } + if s.media != nil { + if upsertErr := s.media.UpsertMedia(*record); upsertErr != nil { + return nil, false, upsertErr + } + } + return record, true, nil +} + +func (s *store) syncArtifactsForJob(record *job) error { + if s.artifacts == nil || record == nil { + return nil + } + return s.artifacts.UpsertArtifacts(record.UserID, record.ProjectID, buildArtifacts(record)) +} + +func (s *store) backfillArtifacts(userID, projectID, jobID string) error { + if s.artifacts == nil { + return nil + } + if jobID != "" { + record, exists, err := s.getJob(userID, jobID) + if err != nil || !exists || record == nil { + return err + } + if projectID != "" && record.ProjectID != projectID { + return nil + } + return s.syncArtifactsForJob(record) + } + + const jobPageSize = 200 + totalJobs := 0 + for jobOffset := 0; ; jobOffset += jobPageSize { + jobs, total, err := s.jobs.ListJobs(userID, "", "", projectID, jobPageSize, jobOffset) + if err != nil { + return err + } + totalJobs = total + for idx := range jobs { + if err := s.syncArtifactsForJob(&jobs[idx]); err != nil { + return err + } + } + if jobOffset+len(jobs) >= totalJobs || len(jobs) == 0 { + break + } + } + return nil +} + func cloneJob(record *job) *job { if record == nil { return nil @@ -371,21 +493,29 @@ func buildArtifacts(record *job) []artifact { storageStatus = "embedded" } } - items = append(items, artifact{ - ID: buildArtifactID(record.JobID, idx), - JobID: record.JobID, - MediaID: strings.TrimSpace(ref.MediaID), - Filename: filename, - ContentType: contentType, - SizeBytes: ref.SizeBytes, - CreatedAt: createdAt, - ExpiresAt: expiresAt, - URL: strings.TrimSpace(ref.URL), - Visibility: strings.TrimSpace(ref.Visibility), - SHA256: strings.TrimSpace(ref.SHA256), - StorageStatus: storageStatus, - Ref: ref, - }) + item := artifact{ + ID: buildArtifactID(record.JobID, idx), + JobID: record.JobID, + ProjectID: record.ProjectID, + MediaID: strings.TrimSpace(ref.MediaID), + Filename: filename, + ContentType: contentType, + SizeBytes: ref.SizeBytes, + CreatedAt: createdAt, + ExpiresAt: expiresAt, + URL: strings.TrimSpace(ref.URL), + Visibility: strings.TrimSpace(ref.Visibility), + SHA256: strings.TrimSpace(ref.SHA256), + StorageStatus: storageStatus, + SourceSkillID: strings.TrimSpace(record.SkillID), + SourceModelID: strings.TrimSpace(record.ModelID), + SourceRouteKey: strings.TrimSpace(record.RouteKey), + SourceInput: cloneMap(record.Input), + PromptText: primaryPromptText(record.Input), + Usage: cloneMap(record.Usage), + Ref: ref, + } + items = append(items, item) } return items } @@ -401,6 +531,18 @@ func buildArtifactIDs(jobID string, refs []resultRef) []string { return ids } +func primaryPromptText(input map[string]any) string { + if input == nil { + return "" + } + return strings.TrimSpace(stringValue( + input["prompt"], + input["motion_prompt"], + input["scene_prompt"], + input["text"], + )) +} + func buildArtifactID(jobID string, idx int) string { return "art_" + strings.TrimPrefix(jobID, "job_") + "_" + strconv.Itoa(idx) } diff --git a/internal/server/types.go b/internal/server/types.go index 8263faa..5853213 100644 --- a/internal/server/types.go +++ b/internal/server/types.go @@ -90,19 +90,26 @@ type logEntry struct { } type artifact struct { - ID string `json:"id"` - JobID string `json:"job_id"` - MediaID string `json:"media_id,omitempty"` - Filename string `json:"filename"` - ContentType string `json:"content_type"` - SizeBytes int64 `json:"size_bytes"` - CreatedAt string `json:"created_at"` - ExpiresAt string `json:"expires_at"` - URL string `json:"url,omitempty"` - Visibility string `json:"visibility,omitempty"` - SHA256 string `json:"sha256,omitempty"` - StorageStatus string `json:"storage_status,omitempty"` - Ref resultRef `json:"-"` + ID string `json:"id"` + JobID string `json:"job_id"` + ProjectID string `json:"project_id,omitempty"` + MediaID string `json:"media_id,omitempty"` + Filename string `json:"filename"` + ContentType string `json:"content_type"` + SizeBytes int64 `json:"size_bytes"` + CreatedAt string `json:"created_at"` + ExpiresAt string `json:"expires_at"` + URL string `json:"url,omitempty"` + Visibility string `json:"visibility,omitempty"` + SHA256 string `json:"sha256,omitempty"` + StorageStatus string `json:"storage_status,omitempty"` + SourceSkillID string `json:"source_skill_id,omitempty"` + SourceModelID string `json:"source_model_id,omitempty"` + SourceRouteKey string `json:"source_route_key,omitempty"` + SourceInput map[string]any `json:"source_input,omitempty"` + PromptText string `json:"prompt_text,omitempty"` + Usage map[string]any `json:"usage,omitempty"` + Ref resultRef `json:"-"` } type media struct { diff --git a/internal/server/video_test.go b/internal/server/video_test.go index 4b51242..91642a6 100644 --- a/internal/server/video_test.go +++ b/internal/server/video_test.go @@ -1,10 +1,12 @@ package server import ( + "bytes" "context" "encoding/base64" "encoding/json" "io" + "mime/multipart" "net/http" "net/http/httptest" "os" @@ -271,6 +273,134 @@ func TestResolveVideoReferencesSupportsImagesArray(t *testing.T) { } } +func TestResolveVideoReferencesAcceptsCanonicalImageURL(t *testing.T) { + cfg := Config{ + SQLitePath: filepath.Join(t.TempDir(), "popiart.db"), + SkillhubDir: makeEmptySkillhub(t), + SessionSecret: "test-secret", + } + server, err := NewWithConfig(cfg) + if err != nil { + t.Fatalf("NewWithConfig: %v", err) + } + + imageBytes := tinyPNG(t) + refSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "image/png") + _, _ = w.Write(imageBytes) + })) + defer refSrv.Close() + + refs, err := server.resolveVideoReferences(context.Background(), &job{ + UserID: "user_test", + UpstreamKey: "sk-upstream", + }, map[string]any{ + "image_url": refSrv.URL + "/single.png", + }) + if err != nil { + t.Fatalf("resolveVideoReferences canonical image_url: %v", err) + } + if len(refs) != 1 { + t.Fatalf("expected 1 ref, got %#v", refs) + } + if refs[0].URL != refSrv.URL+"/single.png" { + t.Fatalf("expected URL to round-trip, got %q", refs[0].URL) + } + if refs[0].ContentType != "image/png" { + t.Fatalf("expected image/png, got %q", refs[0].ContentType) + } + if !bytes.Equal(refs[0].Content, imageBytes) { + t.Fatal("expected canonical image_url bytes to match") + } +} + +func TestResolveVideoReferencesAcceptsSignedSameOriginMediaURL(t *testing.T) { + cfg := Config{ + SQLitePath: filepath.Join(t.TempDir(), "popiart.db"), + SkillhubDir: makeEmptySkillhub(t), + SessionSecret: "test-secret", + } + server, err := NewWithConfig(cfg) + if err != nil { + t.Fatalf("NewWithConfig: %v", err) + } + srv := httptest.NewServer(server.Handler()) + defer srv.Close() + server.cfg.PublicBaseURL = srv.URL + + sessionToken, _, ok, err := server.store.createSession("sk-video-same-origin-user") + if err != nil { + t.Fatalf("createSession: %v", err) + } + if !ok { + t.Fatal("expected session creation to succeed") + } + current, exists, err := server.store.session(sessionToken) + if err != nil { + t.Fatalf("load session: %v", err) + } + if !exists { + t.Fatal("expected stored session") + } + + imageBytes := tinyPNG(t) + var body bytes.Buffer + writer := multipart.NewWriter(&body) + part, err := writer.CreateFormFile("file", "video-reference.png") + if err != nil { + t.Fatalf("create form file: %v", err) + } + if _, err := part.Write(imageBytes); err != nil { + t.Fatalf("write form file: %v", err) + } + if err := writer.Close(); err != nil { + t.Fatalf("close multipart writer: %v", err) + } + + req := httptest.NewRequest(http.MethodPost, "/v1/media/upload", &body) + req.Header.Set("Authorization", "Bearer "+sessionToken) + req.Header.Set("Content-Type", writer.FormDataContentType()) + rec := httptest.NewRecorder() + server.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusCreated { + t.Fatalf("expected media upload 201, got %d body=%s", rec.Code, rec.Body.String()) + } + + var envelope struct { + OK bool `json:"ok"` + Data media `json:"data"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &envelope); err != nil { + t.Fatalf("decode media upload response: %v body=%s", err, rec.Body.String()) + } + if !envelope.OK || envelope.Data.URL == "" { + t.Fatalf("expected signed media url, got %s", rec.Body.String()) + } + + refs, err := server.resolveVideoReferences(context.Background(), &job{ + UserID: current.User.ID, + SessionID: current.Token, + UpstreamKey: current.UpstreamKey, + }, map[string]any{ + "image_url": envelope.Data.URL, + }) + if err != nil { + t.Fatalf("resolveVideoReferences same-origin signed media url: %v", err) + } + if len(refs) != 1 { + t.Fatalf("expected 1 ref, got %#v", refs) + } + if refs[0].URL != envelope.Data.URL { + t.Fatalf("expected signed URL to round-trip, got %q", refs[0].URL) + } + if refs[0].ContentType != "image/png" { + t.Fatalf("expected image/png, got %q", refs[0].ContentType) + } + if !bytes.Equal(refs[0].Content, imageBytes) { + t.Fatal("expected same-origin signed media bytes to match") + } +} + func TestFetchVideoTaskFallsBackToGenericTaskEnvelope(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path {