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 {