From 6d14e54e36ea918036eb5ff9d77ebdd06fee8ad9 Mon Sep 17 00:00:00 2001 From: jiajia Date: Wed, 8 Apr 2026 11:50:21 +0800 Subject: [PATCH] Add stable media URLs for artifacts and uploads --- README.md | 4 +- docs/persistence.md | 33 +++- internal/server/artifact_upload_test.go | 29 ++++ internal/server/media.go | 193 ++++++++++++++++++++++ internal/server/media_test.go | 106 ++++++++++++ internal/server/newapi.go | 107 +++++++++++- internal/server/server.go | 210 +++++++++++++++++++++++- internal/server/store.go | 32 +++- internal/server/types.go | 51 ++++-- internal/server/video_test.go | 63 +++++++ 10 files changed, 789 insertions(+), 39 deletions(-) create mode 100644 internal/server/media.go create mode 100644 internal/server/media_test.go diff --git a/README.md b/README.md index ef8b777..6720b96 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ It is meant to validate these product-facing APIs first: - `run` - `jobs` - `artifacts` +- `media` - basic `budget`, `project`, and `models` stubs ## Project Relationship @@ -61,7 +62,8 @@ In local development, `/tmp/Popiart_skillhub` is used automatically when it exis ## Notes - Local persistence is intentionally thin: `sessions`, `job refs`, and project route overrides live in SQLite. -- `artifact` no longer has a local blob store. For sync image results, the server stores `result_refs_json` in the job row and derives artifact metadata on demand. +- The local backend now keeps a lightweight media store under `POPIART_DATA_DIR/media/` and exposes stable `media` URLs for uploaded files and persisted job outputs. +- `artifact` metadata is still derived from `result_refs_json`, but new artifacts bind to a local `media_id` and a stable `url`. - `job logs` are synthesized from job state transitions instead of being stored as a separate table. - The local development backend verifies your login key against the local `PopiNewAPI`. - This server is a development backend, not the final production architecture. diff --git a/docs/persistence.md b/docs/persistence.md index bd2ce5a..f940023 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -14,7 +14,6 @@ 所以 `popiartServer` 不再重复持有： -- 本地 artifact blob store - 本地 artifact metadata table - 本地 job_logs table @@ -23,6 +22,7 @@ - `sessions` - `jobs`（更准确说是 job refs） - `skill_routes` +- `media blobs + metadata` ## 当前关系 @@ -30,14 +30,16 @@ session -> jobs jobs -> result_refs_json project -> skill_routes +media -> local files + metadata json ``` 其中： - `session` 保存 PopiArt 登录态和对应的 `PopiNewAPI token` - `jobs` 保存 skill 语义、用户归属、上游引用和同步结果引用 -- `result_refs_json` 保存同步能力的返回引用，例如 `data_url` 或远端 `url` +- `result_refs_json` 保存同步能力的结果引用；新结果会优先 re-host 到本地 media store - `skill_routes` 保存项目级路由覆盖 +- `media` 保存稳定 URL 所需的本地文件与元数据 ## 存储位置 @@ -66,7 +68,10 @@ POPIART_SQLITE_PATH=./data/popiart.db 2. `JobRepository` 3. `RouteRepository` -没有单独的 blob storage 抽象。 +本地开发版没有对象存储依赖，但现在有一个轻量 media 存储层： + +- blob 文件保存在 `POPIART_DATA_DIR/media/blobs/` +- metadata JSON 保存在 `POPIART_DATA_DIR/media/meta/` ## SQLite Schema @@ -141,15 +146,24 @@ CREATE TABLE skill_routes ( ### 完成同步结果 job 1. 如果 `PopiNewAPI` 直接返回同步结果，例如 `b64_json` 或 `url` -2. `popiartServer` 只把它转换成 `result_refs_json` -3. `GET /jobs/:id/artifacts` 时再从 `result_refs_json` 派生 artifact 列表 +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 列表 ### 拉取 artifact 1. server 从 `artifact_id` 反推出 `job_id + result index` 2. 读取 `jobs.result_refs_json` -3. 如果是 `data_url`，直接解码并流式返回 -4. 如果是远端 `url`，server 代理下载并返回 +3. 如果是本地 `local_path`，直接读取本地文件并流式返回 +4. 如果是旧的 `data_url`，直接解码并流式返回 +5. 如果是旧的远端 `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 的稳定内容路径 ## 当前边界 @@ -157,7 +171,8 @@ CREATE TABLE skill_routes ( - `session` 会保留 - `job ref` 会保留 - `artifact` 通过 `result_refs_json` 继续可读 + - 新 `media` 文件和 metadata 会继续可读 - 已存在的 `pending/running` job 仍然不会自动恢复执行 - 视频类 `upstream_task` 路径只预留了字段，后续再接 `PopiNewAPI` task 查询 -- 对同步图像结果来说，`data_url` 仍然会落在 SQLite 的 `result_refs_json` - 这是当前 `PopiNewAPI` 没有统一 file id 的现实折中，但已经不再有本地 artifact 表和 blob store +- 旧数据里仍可能存在 `data_url` 或上游 `url` +- 新写入路径优先落本地 media store，从而给 artifact 补出稳定 `url` diff --git a/internal/server/artifact_upload_test.go b/internal/server/artifact_upload_test.go index 7b69eba..e0ef795 100644 --- a/internal/server/artifact_upload_test.go +++ b/internal/server/artifact_upload_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "io" "mime/multipart" "net/http" "net/http/httptest" @@ -21,6 +22,9 @@ func TestArtifactUploadCreatesReadableArtifactForSourceArtifactID(t *testing.T) 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-upload-user") if err != nil { @@ -80,9 +84,34 @@ func TestArtifactUploadCreatesReadableArtifactForSourceArtifactID(t *testing.T) if envelope.Data.ID == "" { t.Fatalf("expected artifact id, got %#v", envelope.Data) } + if envelope.Data.MediaID == "" { + t.Fatalf("expected media id, got %#v", envelope.Data) + } if envelope.Data.ContentType != "image/png" { t.Fatalf("expected image/png, got %q", envelope.Data.ContentType) } + if envelope.Data.URL == "" { + t.Fatalf("expected stable artifact url, got %#v", envelope.Data) + } + if envelope.Data.StorageStatus != "ready" { + t.Fatalf("expected storage status ready, got %#v", envelope.Data.StorageStatus) + } + + contentResp, err := http.Get(envelope.Data.URL) + if err != nil { + t.Fatalf("GET media content: %v", err) + } + defer contentResp.Body.Close() + if contentResp.StatusCode != http.StatusOK { + t.Fatalf("expected media content 200, got %d", contentResp.StatusCode) + } + streamed, err := io.ReadAll(contentResp.Body) + if err != nil { + t.Fatalf("read media content: %v", err) + } + if !bytes.Equal(streamed, imageBytes) { + t.Fatal("expected stable media url to serve uploaded bytes") + } editRef, err := server.resolveImageToImageReference(context.Background(), &job{ UserID: current.User.ID, diff --git a/internal/server/media.go b/internal/server/media.go new file mode 100644 index 0000000..fcbc801 --- /dev/null +++ b/internal/server/media.go @@ -0,0 +1,193 @@ +package server + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/url" + "os" + "path/filepath" + "strings" + "time" +) + +type mediaRecord struct { + media + UserID string `json:"user_id"` + LocalPath string `json:"local_path"` +} + +func mediaDir(cfg Config) string { + return filepath.Join(cfg.DataDir, "media") +} + +func mediaBlobDir(cfg Config) string { + return filepath.Join(mediaDir(cfg), "blobs") +} + +func mediaMetaDir(cfg Config) string { + return filepath.Join(mediaDir(cfg), "meta") +} + +func mediaMetaPath(cfg Config, mediaID string) string { + return filepath.Join(mediaMetaDir(cfg), mediaID+".json") +} + +func mediaBlobPath(cfg Config, mediaID, filename string) string { + ext := filepath.Ext(filename) + if ext == "" { + ext = extensionFromContentType("") + } + return filepath.Join(mediaBlobDir(cfg), mediaID+ext) +} + +func publicBaseURL(cfg Config) string { + base := strings.TrimSpace(cfg.PublicBaseURL) + if base == "" { + base = strings.TrimSpace(os.Getenv("POPIART_SERVER_ADDR")) + } + if base == "" { + base = "127.0.0.1:8080" + } + if !strings.HasPrefix(base, "http://") && !strings.HasPrefix(base, "https://") { + base = "http://" + base + } + base = strings.Replace(base, "0.0.0.0", "127.0.0.1", 1) + return strings.TrimRight(base, "/") +} + +func mediaContentURL(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) { + if len(content) == 0 { + return mediaRecord{}, fmt.Errorf("media content is required") + } + filename = sanitizeFilename(strings.TrimSpace(filename)) + if filename == "" || filename == "artifact" { + filename = "artifact" + extensionFromContentType(contentType) + } + contentType = strings.TrimSpace(contentType) + if contentType == "" { + contentType = "application/octet-stream" + } + visibility = strings.TrimSpace(visibility) + if visibility == "" { + visibility = "unlisted" + } + mediaID = strings.TrimSpace(mediaID) + if mediaID == "" { + mediaID = "med_" + randomID() + } + if err := os.MkdirAll(mediaBlobDir(cfg), 0o755); err != nil { + return mediaRecord{}, err + } + if err := os.MkdirAll(mediaMetaDir(cfg), 0o755); err != nil { + return mediaRecord{}, err + } + + sum := sha256.Sum256(content) + blobPath := mediaBlobPath(cfg, mediaID, filename) + if err := os.WriteFile(blobPath, content, 0o644); err != nil { + return mediaRecord{}, err + } + + record := mediaRecord{ + media: media{ + ID: mediaID, + ArtifactID: strings.TrimSpace(artifactID), + ProjectID: strings.TrimSpace(projectID), + Filename: filename, + ContentType: contentType, + SizeBytes: int64(len(content)), + CreatedAt: time.Now().UTC().Format(time.RFC3339), + URL: mediaContentURL(cfg, mediaID), + Visibility: visibility, + SHA256: hex.EncodeToString(sum[:]), + }, + UserID: strings.TrimSpace(userID), + LocalPath: blobPath, + } + + metaBytes, err := json.Marshal(record) + if err != nil { + return mediaRecord{}, err + } + if err := os.WriteFile(mediaMetaPath(cfg, mediaID), metaBytes, 0o644); err != nil { + return mediaRecord{}, err + } + return record, nil +} + +func loadMediaRecord(cfg Config, mediaID string) (*mediaRecord, bool, error) { + mediaID = strings.TrimSpace(mediaID) + if mediaID == "" { + return nil, false, nil + } + data, err := os.ReadFile(mediaMetaPath(cfg, mediaID)) + if os.IsNotExist(err) { + return nil, false, nil + } + if err != nil { + return nil, false, err + } + var record mediaRecord + if err := json.Unmarshal(data, &record); err != nil { + return nil, false, err + } + return &record, true, nil +} + +func (s *Server) persistResultRefs(ctx context.Context, record *job, refs []resultRef) ([]resultRef, error) { + if record == nil || len(refs) == 0 { + return refs, nil + } + items := make([]resultRef, 0, len(refs)) + for idx, ref := range refs { + contentType, _, reader, err := s.newapi.openResultRef(ctx, record.UpstreamKey, ref) + if err != nil { + return nil, err + } + content, readErr := io.ReadAll(reader) + reader.Close() + if readErr != nil { + return nil, readErr + } + filename := strings.TrimSpace(ref.Filename) + if filename == "" { + filename = inferFilenameFromRef(ref, buildArtifactID(record.JobID, idx)) + } + artifactID := buildArtifactID(record.JobID, idx) + mediaRecord, err := persistMediaContent( + s.cfg, + record.UserID, + record.ProjectID, + artifactID, + filename, + defaultString(strings.TrimSpace(contentType), strings.TrimSpace(ref.ContentType)), + defaultString(strings.TrimSpace(ref.Visibility), "unlisted"), + content, + "", + ) + if err != nil { + return nil, err + } + items = append(items, resultRef{ + Kind: "local_path", + URL: mediaRecord.URL, + LocalPath: mediaRecord.LocalPath, + MediaID: mediaRecord.ID, + Filename: mediaRecord.Filename, + ContentType: mediaRecord.ContentType, + SizeBytes: mediaRecord.SizeBytes, + Visibility: mediaRecord.Visibility, + SHA256: mediaRecord.SHA256, + StorageStatus: "ready", + }) + } + return items, nil +} diff --git a/internal/server/media_test.go b/internal/server/media_test.go new file mode 100644 index 0000000..7a2a48c --- /dev/null +++ b/internal/server/media_test.go @@ -0,0 +1,106 @@ +package server + +import ( + "bytes" + "encoding/json" + "io" + "mime/multipart" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" +) + +func TestMediaUploadGetAndContent(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-user") + 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) + if err := writer.WriteField("project_id", "proj_media_demo"); err != nil { + t.Fatalf("write project_id field: %v", err) + } + if err := writer.WriteField("visibility", "public"); err != nil { + t.Fatalf("write visibility field: %v", err) + } + part, err := writer.CreateFormFile("file", "poster.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 status 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 upload response: %v body=%s", err, rec.Body.String()) + } + if !envelope.OK { + t.Fatalf("expected ok upload response, got %s", rec.Body.String()) + } + if envelope.Data.ID == "" { + t.Fatalf("expected media id, got %#v", envelope.Data) + } + if envelope.Data.URL == "" { + t.Fatalf("expected stable url, got %#v", envelope.Data) + } + + 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, got %d body=%s", getRec.Code, getRec.Body.String()) + } + + contentResp, err := http.Get(envelope.Data.URL) + if err != nil { + t.Fatalf("GET media content: %v", err) + } + defer contentResp.Body.Close() + if contentResp.StatusCode != http.StatusOK { + t.Fatalf("expected media content 200, got %d", contentResp.StatusCode) + } + content, err := io.ReadAll(contentResp.Body) + if err != nil { + t.Fatalf("read content body: %v", err) + } + if !bytes.Equal(content, imageBytes) { + t.Fatal("expected media content bytes to match uploaded file") + } +} diff --git a/internal/server/newapi.go b/internal/server/newapi.go index 9be4d52..25a5807 100644 --- a/internal/server/newapi.go +++ b/internal/server/newapi.go @@ -30,6 +30,7 @@ type Config struct { NewAPIToken string DefaultImageModel string DefaultVideoModel string + PublicBaseURL string DataDir string SQLitePath string SessionSecret string @@ -42,11 +43,16 @@ func ConfigFromEnv() Config { NewAPIToken: strings.TrimSpace(os.Getenv("POPIART_NEWAPI_TOKEN")), DefaultImageModel: strings.TrimSpace(os.Getenv("POPIART_DEFAULT_IMAGE_MODEL")), DefaultVideoModel: strings.TrimSpace(os.Getenv("POPIART_DEFAULT_VIDEO_MODEL")), + PublicBaseURL: strings.TrimSpace(os.Getenv("POPIART_PUBLIC_BASE_URL")), DataDir: strings.TrimSpace(os.Getenv("POPIART_DATA_DIR")), SQLitePath: strings.TrimSpace(os.Getenv("POPIART_SQLITE_PATH")), SessionSecret: strings.TrimSpace(os.Getenv("POPIART_SESSION_SECRET")), SkillhubDir: strings.TrimSpace(os.Getenv("POPIART_SKILLHUB_DIR")), } + return normalizeConfig(cfg) +} + +func normalizeConfig(cfg Config) Config { if cfg.NewAPIBaseURL == "" { cfg.NewAPIBaseURL = "http://127.0.0.1:3000" } @@ -57,7 +63,11 @@ func ConfigFromEnv() Config { cfg.DefaultVideoModel = "viduq2" } if cfg.DataDir == "" { - cfg.DataDir = "./data" + if strings.TrimSpace(cfg.SQLitePath) != "" { + cfg.DataDir = filepath.Dir(cfg.SQLitePath) + } else { + cfg.DataDir = "./data" + } } if cfg.SQLitePath == "" { cfg.SQLitePath = filepath.Join(cfg.DataDir, "popiart.db") @@ -131,6 +141,7 @@ type imageEditReference struct { Filename string ContentType string Content []byte + URL string } type openAIVideoResponse struct { @@ -587,7 +598,7 @@ func (c *newAPIClient) generateGeminiImageRefs(ctx context.Context, token, model "generationConfig": generationConfig, } if len(imageConfig) > 0 { - payload["imageConfig"] = imageConfig + generationConfig["imageConfig"] = imageConfig } body, err := json.Marshal(payload) @@ -629,6 +640,9 @@ func (c *newAPIClient) submitImageToVideoTask(ctx context.Context, token, modelI if prompt == "" { prompt = "Generate a short polished image-to-video clip from the provided reference image." } + if strings.HasPrefix(strings.ToLower(strings.TrimSpace(modelID)), "vidu") && strings.TrimSpace(ref.URL) != "" { + return c.submitImageToVideoTaskByURL(ctx, token, modelID, input, ref, prompt) + } if len(ref.Content) == 0 { return "", errors.New("reference image content is required") } @@ -713,6 +727,71 @@ func (c *newAPIClient) submitImageToVideoTask(ctx context.Context, token, modelI return taskID, nil } +func (c *newAPIClient) submitImageToVideoTaskByURL(ctx context.Context, token, modelID string, input map[string]any, ref imageEditReference, prompt string) (string, error) { + payload := map[string]any{ + "model": modelID, + "prompt": prompt, + "images": []string{strings.TrimSpace(ref.URL)}, + "duration": 5, + } + if duration := strings.TrimSpace(resolveVideoDurationSeconds(input)); duration != "" { + if parsed, err := strconv.Atoi(duration); err == nil && parsed > 0 { + payload["duration"] = parsed + } + } + if size := resolveVideoSize(modelID, input, ref); size != "" { + payload["size"] = size + } + if aspectRatio := resolveVideoAspectRatio(input, ref); aspectRatio != "" { + payload["metadata"] = map[string]any{ + "aspect_ratio": aspectRatio, + } + } + + body, err := json.Marshal(payload) + if err != nil { + return "", err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/v1/videos", bytes.NewReader(body)) + if err != nil { + return "", err + } + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + + resp, err := c.httpClient.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return "", err + } + + var decoded openAIVideoResponse + if err := json.Unmarshal(respBody, &decoded); err != nil { + return "", fmt.Errorf("decode PopiNewAPI video submit response: %w", err) + } + if resp.StatusCode >= 400 { + return "", decodeTaskAPIError(respBody, resp.StatusCode) + } + if decoded.Error != nil && strings.TrimSpace(decoded.Error.Message) != "" { + return "", errors.New(decoded.Error.Message) + } + + taskID := strings.TrimSpace(decoded.ID) + if taskID == "" { + taskID = strings.TrimSpace(decoded.TaskID) + } + if taskID == "" { + return "", errors.New("PopiNewAPI returned no task id") + } + return taskID, nil +} + func (c *newAPIClient) fetchVideoTask(ctx context.Context, token, taskID string) (*videoTaskResult, error) { if !c.enabled() { return nil, errors.New("PopiNewAPI base URL is not configured") @@ -997,6 +1076,12 @@ func resolveGeminiImageSize(input map[string]any) string { func (c *newAPIClient) openResultRef(ctx context.Context, token string, ref resultRef) (string, int64, io.ReadCloser, error) { switch ref.Kind { + case "local_path": + file, err := os.Open(ref.LocalPath) + if err != nil { + return "", 0, nil, err + } + return defaultString(strings.TrimSpace(ref.ContentType), "application/octet-stream"), ref.SizeBytes, file, nil case "data_url": contentType, content, err := decodeDataURL(ref.DataURL) if err != nil { @@ -1137,6 +1222,24 @@ func resolveVideoSize(modelID string, input map[string]any, ref imageEditReferen return size } +func resolveVideoAspectRatio(input map[string]any, ref imageEditReference) string { + if input != nil { + if aspectRatio := strings.TrimSpace(stringValue(input["aspect_ratio"])); aspectRatio != "" { + return aspectRatio + } + } + if len(ref.Content) > 0 { + cfg, _, err := image.DecodeConfig(bytes.NewReader(ref.Content)) + if err == nil && cfg.Width > 0 && cfg.Height > 0 { + if cfg.Width >= cfg.Height { + return "16:9" + } + return "9:16" + } + } + return "" +} + func resolveBaseVideoSize(input map[string]any, ref imageEditReference) string { if input != nil { if size := strings.TrimSpace(stringValue(input["size"])); size != "" { diff --git a/internal/server/server.go b/internal/server/server.go index 20dd7de..82173f1 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -9,6 +9,7 @@ import ( "io" "log" "net/http" + "os" "path/filepath" "strconv" "strings" @@ -31,6 +32,7 @@ func New() *Server { } func NewWithConfig(cfg Config) (*Server, error) { + cfg = normalizeConfig(cfg) store, err := newStore(cfg) if err != nil { return nil, err @@ -59,6 +61,8 @@ func (s *Server) routes() { s.mux.HandleFunc("/v1/skills/", s.handleSkill) s.mux.HandleFunc("/v1/jobs", s.handleJobs) 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/upload", s.handleArtifactUpload) s.mux.HandleFunc("/v1/artifacts/", s.handleArtifact) s.mux.HandleFunc("/v1/budget", s.handleBudget) @@ -468,6 +472,7 @@ func (s *Server) handleArtifactUpload(w http.ResponseWriter, r *http.Request) { filename = "artifact" + extensionFromContentType(contentType) } filename = sanitizeFilename(filename) + visibility := strings.TrimSpace(r.FormValue("visibility")) input := map[string]any{ "filename": filename, @@ -499,12 +504,33 @@ func (s *Server) handleArtifactUpload(w http.ResponseWriter, r *http.Request) { return } + refMedia, err := persistMediaContent( + s.cfg, + current.User.ID, + projectID, + buildArtifactID(record.JobID, 0), + filename, + contentType, + visibility, + content, + "", + ) + if err != nil { + writeInternalError(w, "failed to persist upload media", err) + return + } + ref := resultRef{ - Kind: "data_url", - DataURL: buildDataURL(contentType, content), - Filename: filename, - ContentType: contentType, - SizeBytes: int64(len(content)), + Kind: "local_path", + URL: refMedia.URL, + LocalPath: refMedia.LocalPath, + MediaID: refMedia.ID, + Filename: refMedia.Filename, + ContentType: refMedia.ContentType, + SizeBytes: refMedia.SizeBytes, + Visibility: refMedia.Visibility, + SHA256: refMedia.SHA256, + StorageStatus: "ready", } if err := s.store.completeJobWithResults(record.JobID, []resultRef{ref}, nil); err != nil { writeInternalError(w, "failed to persist upload artifact", err) @@ -523,6 +549,140 @@ func (s *Server) handleArtifactUpload(w http.ResponseWriter, r *http.Request) { writeData(w, http.StatusCreated, item) } +func (s *Server) handleMediaUpload(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + methodNotAllowed(w) + return + } + current, ok := s.authenticateSession(w, r) + if !ok { + return + } + + if err := r.ParseMultipartForm(32 << 20); err != nil { + writeError(w, http.StatusBadRequest, "VALIDATION_ERROR", "failed to parse multipart form", map[string]any{ + "details": err.Error(), + }) + return + } + + file, header, err := r.FormFile("file") + if err != nil { + writeError(w, http.StatusBadRequest, "VALIDATION_ERROR", "file is required", nil) + return + } + defer file.Close() + + content, err := io.ReadAll(file) + if err != nil { + writeError(w, http.StatusBadRequest, "BAD_REQUEST", "failed to read upload file", map[string]any{ + "details": err.Error(), + }) + return + } + if len(content) == 0 { + writeError(w, http.StatusBadRequest, "VALIDATION_ERROR", "uploaded file is empty", nil) + return + } + + metadataJSON := strings.TrimSpace(r.FormValue("metadata_json")) + if metadataJSON != "" { + var metadata any + if err := json.Unmarshal([]byte(metadataJSON), &metadata); err != nil { + writeError(w, http.StatusBadRequest, "VALIDATION_ERROR", "metadata_json must be valid JSON", map[string]any{ + "details": err.Error(), + }) + return + } + } + + filename := strings.TrimSpace(r.FormValue("filename")) + if filename == "" { + filename = strings.TrimSpace(header.Filename) + } + contentType := strings.TrimSpace(r.FormValue("content_type")) + if contentType == "" { + contentType = strings.TrimSpace(header.Header.Get("Content-Type")) + } + if contentType == "" || contentType == "application/octet-stream" { + contentType = http.DetectContentType(content) + } + if filename == "" { + filename = "media" + extensionFromContentType(contentType) + } + filename = sanitizeFilename(filename) + + record, err := persistMediaContent( + s.cfg, + current.User.ID, + strings.TrimSpace(r.FormValue("project_id")), + "", + filename, + contentType, + strings.TrimSpace(r.FormValue("visibility")), + content, + "", + ) + if err != nil { + writeInternalError(w, "failed to persist media", err) + return + } + writeData(w, http.StatusCreated, record.media) +} + +func (s *Server) handleMedia(w http.ResponseWriter, r *http.Request) { + path := strings.TrimPrefix(r.URL.Path, "/v1/media/") + parts := strings.Split(strings.Trim(path, "/"), "/") + if len(parts) == 0 || parts[0] == "" { + notFound(w) + return + } + + record, exists, err := loadMediaRecord(s.cfg, parts[0]) + if err != nil { + writeInternalError(w, "failed to load media", err) + return + } + if !exists || record == nil { + notFound(w) + return + } + + if len(parts) == 2 && parts[1] == "content" && r.Method == http.MethodGet { + file, err := os.Open(record.LocalPath) + if err != nil { + writeInternalError(w, "failed to read media content", err) + return + } + defer file.Close() + + w.Header().Set("Content-Type", defaultString(record.ContentType, "application/octet-stream")) + if record.SizeBytes > 0 { + w.Header().Set("Content-Length", strconv.FormatInt(record.SizeBytes, 10)) + } + w.WriteHeader(http.StatusOK) + if _, err := io.Copy(w, file); err != nil { + log.Printf("popiartServer: streaming media %s failed: %v", record.ID, err) + } + return + } + + current, ok := s.authenticateSession(w, r) + if !ok { + return + } + if current.User.ID != record.UserID { + notFound(w) + return + } + if len(parts) == 1 && r.Method == http.MethodGet { + writeData(w, http.StatusOK, record.media) + return + } + + notFound(w) +} + func (s *Server) handleArtifact(w http.ResponseWriter, r *http.Request) { current, ok := s.authenticateSession(w, r) if !ok { @@ -958,6 +1118,17 @@ func (s *Server) executeTextToImageJob(record *job) { return } + refs, err = s.persistResultRefs(ctx, record, refs) + if err != nil { + if repoErr := s.store.failJob(record.JobID, "MEDIA_PERSIST_FAILED", "failed to persist generated media", map[string]any{ + "details": err.Error(), + "model_id": modelID, + }); repoErr != nil { + log.Printf("popiartServer: fail job %s failed: %v", record.JobID, repoErr) + } + return + } + if err := s.store.completeJobWithResults(record.JobID, refs, usage); err != nil { log.Printf("popiartServer: complete job %s failed: %v", record.JobID, err) if failErr := s.store.failJob(record.JobID, "RESULT_REF_PERSIST_FAILED", "failed to persist job result refs", map[string]any{ @@ -1010,6 +1181,18 @@ func (s *Server) executeImageToImageJob(record *job) { return } + refs, err = s.persistResultRefs(ctx, record, refs) + if err != nil { + if repoErr := s.store.failJob(record.JobID, "MEDIA_PERSIST_FAILED", "failed to persist generated media", map[string]any{ + "details": err.Error(), + "model_id": modelID, + "route_key": record.RouteKey, + }); repoErr != nil { + log.Printf("popiartServer: fail job %s failed: %v", record.JobID, repoErr) + } + return + } + if err := s.store.completeJobWithResults(record.JobID, refs, usage); err != nil { log.Printf("popiartServer: complete job %s failed: %v", record.JobID, err) if failErr := s.store.failJob(record.JobID, "RESULT_REF_PERSIST_FAILED", "failed to persist job result refs", map[string]any{ @@ -1097,6 +1280,21 @@ func (s *Server) executeImageToVideoJob(record *job) { usage["format"] = taskResult.Format } + persistCtx, cancelPersist := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancelPersist() + refs, err = s.persistResultRefs(persistCtx, record, refs) + if err != nil { + if repoErr := s.store.failJob(record.JobID, "MEDIA_PERSIST_FAILED", "failed to persist generated media", map[string]any{ + "details": err.Error(), + "model_id": modelID, + "route_key": record.RouteKey, + "newapi_task_id": upstreamTaskID, + }); repoErr != nil { + log.Printf("popiartServer: fail job %s failed: %v", record.JobID, repoErr) + } + return + } + if err := s.store.completeJobWithResults(record.JobID, refs, usage); err != nil { log.Printf("popiartServer: complete job %s failed: %v", record.JobID, err) if failErr := s.store.failJob(record.JobID, "RESULT_REF_PERSIST_FAILED", "failed to persist job result refs", map[string]any{ @@ -1168,6 +1366,7 @@ func (s *Server) resolveImageToImageReference(ctx context.Context, record *job, Filename: item.Filename, ContentType: defaultString(contentType, item.ContentType), Content: content, + URL: item.URL, }, nil } @@ -1237,6 +1436,7 @@ func (s *Server) downloadReferenceImage(ctx context.Context, rawURL string) (ima Filename: filenameFromURL(rawURL, "reference"+extensionFromContentType(contentType)), ContentType: contentType, Content: content, + URL: strings.TrimSpace(rawURL), }, nil } diff --git a/internal/server/store.go b/internal/server/store.go index a07bdcb..23e0837 100644 --- a/internal/server/store.go +++ b/internal/server/store.go @@ -360,15 +360,31 @@ func buildArtifacts(record *job) []artifact { filename = inferFilenameFromRef(ref, buildArtifactID(record.JobID, idx)) } contentType := defaultString(strings.TrimSpace(ref.ContentType), "application/octet-stream") + storageStatus := strings.TrimSpace(ref.StorageStatus) + if storageStatus == "" { + switch { + case ref.LocalPath != "": + storageStatus = "ready" + case ref.URL != "": + storageStatus = "upstream" + case ref.DataURL != "": + storageStatus = "embedded" + } + } items = append(items, artifact{ - ID: buildArtifactID(record.JobID, idx), - JobID: record.JobID, - Filename: filename, - ContentType: contentType, - SizeBytes: ref.SizeBytes, - CreatedAt: createdAt, - ExpiresAt: expiresAt, - Ref: ref, + 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, }) } return items diff --git a/internal/server/types.go b/internal/server/types.go index 1ec0c8e..8263faa 100644 --- a/internal/server/types.go +++ b/internal/server/types.go @@ -46,12 +46,17 @@ type jobError struct { } type resultRef struct { - Kind string `json:"kind"` - URL string `json:"url,omitempty"` - DataURL string `json:"data_url,omitempty"` - Filename string `json:"filename,omitempty"` - ContentType string `json:"content_type,omitempty"` - SizeBytes int64 `json:"size_bytes,omitempty"` + Kind string `json:"kind"` + URL string `json:"url,omitempty"` + DataURL string `json:"data_url,omitempty"` + LocalPath string `json:"local_path,omitempty"` + MediaID string `json:"media_id,omitempty"` + Filename string `json:"filename,omitempty"` + ContentType string `json:"content_type,omitempty"` + SizeBytes int64 `json:"size_bytes,omitempty"` + Visibility string `json:"visibility,omitempty"` + SHA256 string `json:"sha256,omitempty"` + StorageStatus string `json:"storage_status,omitempty"` } type job struct { @@ -85,14 +90,32 @@ type logEntry struct { } type artifact struct { - ID string `json:"id"` - JobID string `json:"job_id"` - Filename string `json:"filename"` - ContentType string `json:"content_type"` - SizeBytes int64 `json:"size_bytes"` - CreatedAt string `json:"created_at"` - ExpiresAt string `json:"expires_at"` - Ref resultRef `json:"-"` + 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:"-"` +} + +type media struct { + ID string `json:"id"` + ArtifactID string `json:"artifact_id,omitempty"` + ProjectID string `json:"project_id,omitempty"` + Filename string `json:"filename"` + ContentType string `json:"content_type"` + SizeBytes int64 `json:"size_bytes"` + CreatedAt string `json:"created_at"` + URL string `json:"url"` + Visibility string `json:"visibility,omitempty"` + SHA256 string `json:"sha256,omitempty"` } type project struct { diff --git a/internal/server/video_test.go b/internal/server/video_test.go index d324416..1ebae01 100644 --- a/internal/server/video_test.go +++ b/internal/server/video_test.go @@ -3,6 +3,7 @@ package server import ( "context" "encoding/base64" + "encoding/json" "io" "net/http" "net/http/httptest" @@ -85,6 +86,58 @@ func TestSubmitImageToVideoTaskUsesMultipartInputReference(t *testing.T) { } } +func TestSubmitImageToVideoTaskUsesURLImagesForViduModels(t *testing.T) { + var gotBody map[string]any + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/videos" { + http.NotFound(w, r) + return + } + if got := r.Header.Get("Content-Type"); got != "application/json" { + t.Fatalf("expected application/json content type, got %q", got) + } + if err := json.NewDecoder(r.Body).Decode(&gotBody); err != nil { + t.Fatalf("decode request body: %v", err) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"task_vidu_url_123","status":"queued"}`)) + })) + defer srv.Close() + + client := newNewAPIClient(Config{NewAPIBaseURL: srv.URL}) + taskID, err := client.submitImageToVideoTask(context.Background(), "sk-test", "viduq2-pro-fast", map[string]any{ + "prompt": "animate this still gently", + "duration_s": 5, + "aspect_ratio": "9:16", + }, imageEditReference{ + Filename: "reference.png", + ContentType: "image/png", + URL: "https://media.popi.test/m/demo/reference.png", + }) + if err != nil { + t.Fatalf("submitImageToVideoTask: %v", err) + } + + if taskID != "task_vidu_url_123" { + t.Fatalf("expected task id task_vidu_url_123, got %q", taskID) + } + if gotBody["model"] != "viduq2-pro-fast" { + t.Fatalf("unexpected model: %#v", gotBody["model"]) + } + images, ok := gotBody["images"].([]any) + if !ok || len(images) != 1 || images[0] != "https://media.popi.test/m/demo/reference.png" { + t.Fatalf("expected one image url, got %#v", gotBody["images"]) + } + if gotBody["duration"] != float64(5) { + t.Fatalf("unexpected duration: %#v", gotBody["duration"]) + } + metadata, ok := gotBody["metadata"].(map[string]any) + if !ok || metadata["aspect_ratio"] != "9:16" { + t.Fatalf("unexpected metadata: %#v", gotBody["metadata"]) + } +} + func TestFetchVideoTaskFallsBackToGenericTaskEnvelope(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { @@ -176,6 +229,7 @@ func TestExecuteImageToVideoJobCompletesAndArtifactCanBeRead(t *testing.T) { if err != nil { t.Fatalf("NewWithConfig: %v", err) } + server.cfg.PublicBaseURL = "http://127.0.0.1:8080" token, _, ok, err := server.store.createSession("sk-test-upstream") if err != nil { @@ -241,6 +295,15 @@ func TestExecuteImageToVideoJobCompletesAndArtifactCanBeRead(t *testing.T) { if item.ContentType != "video/mp4" { t.Fatalf("expected video/mp4 artifact, got %q", item.ContentType) } + if item.MediaID == "" { + t.Fatalf("expected media id on persisted artifact, got %#v", item) + } + if item.URL == "" { + t.Fatalf("expected stable url on persisted artifact, got %#v", item) + } + if item.StorageStatus != "ready" { + t.Fatalf("expected ready storage status, got %#v", item.StorageStatus) + } if !strings.HasSuffix(item.Filename, ".mp4") { t.Fatalf("expected .mp4 filename, got %q", item.Filename) }