You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1725 lines
47 KiB
1725 lines
47 KiB
package server
|
|
|
|
import (
|
|
"context"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"log"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
type Server struct {
|
|
store *store
|
|
mux *http.ServeMux
|
|
cfg Config
|
|
newapi *newAPIClient
|
|
}
|
|
|
|
func New() *Server {
|
|
svc, err := NewWithConfig(ConfigFromEnv())
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
return svc
|
|
}
|
|
|
|
func NewWithConfig(cfg Config) (*Server, error) {
|
|
cfg = normalizeConfig(cfg)
|
|
store, err := newStore(cfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
s := &Server{
|
|
store: store,
|
|
mux: http.NewServeMux(),
|
|
cfg: cfg,
|
|
newapi: newNewAPIClient(cfg),
|
|
}
|
|
s.routes()
|
|
return s, nil
|
|
}
|
|
|
|
func (s *Server) Handler() http.Handler {
|
|
return s.mux
|
|
}
|
|
|
|
func (s *Server) routes() {
|
|
s.mux.HandleFunc("/health", s.handleHealth)
|
|
s.mux.HandleFunc("/v1/auth/login", s.handleAuthLogin)
|
|
s.mux.HandleFunc("/v1/auth/me", s.handleAuthMe)
|
|
s.mux.HandleFunc("/v1/auth/logout", s.handleAuthLogout)
|
|
s.mux.HandleFunc("/v1/auth/token/rotate", s.handleAuthTokenRotate)
|
|
s.mux.HandleFunc("/v1/skills", s.handleSkills)
|
|
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)
|
|
s.mux.HandleFunc("/v1/budget/usage", s.handleBudgetUsage)
|
|
s.mux.HandleFunc("/v1/budget/limits", s.handleBudgetLimits)
|
|
s.mux.HandleFunc("/v1/projects", s.handleProjects)
|
|
s.mux.HandleFunc("/v1/projects/", s.handleProject)
|
|
s.mux.HandleFunc("/v1/models", s.handleModels)
|
|
s.mux.HandleFunc("/v1/models/routes", s.handleModelRoutes)
|
|
s.mux.HandleFunc("/v1/models/routes/overrides", s.handleModelRouteOverrides)
|
|
s.mux.HandleFunc("/v1/models/routes/overrides/unset", s.handleModelRouteOverrideUnset)
|
|
s.mux.HandleFunc("/v1/models/infer", s.handleModelsInfer)
|
|
}
|
|
|
|
func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
|
|
writeJSON(w, http.StatusOK, map[string]any{
|
|
"ok": true,
|
|
"service": "popiart-server-dev",
|
|
"version": "0.2.0",
|
|
"time": time.Now().UTC().Format(time.RFC3339),
|
|
})
|
|
}
|
|
|
|
func (s *Server) handleAuthLogin(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
var req struct {
|
|
Key string `json:"key"`
|
|
}
|
|
if !decodeJSON(w, r, &req) {
|
|
return
|
|
}
|
|
req.Key = strings.TrimSpace(req.Key)
|
|
if req.Key == "" {
|
|
writeError(w, http.StatusUnauthorized, "UNAUTHENTICATED", "key is required", nil)
|
|
return
|
|
}
|
|
if s.newapi == nil || !s.newapi.enabled() {
|
|
writeError(w, http.StatusInternalServerError, "SERVER_ERROR", "PopiNewAPI base URL is not configured", nil)
|
|
return
|
|
}
|
|
ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
|
|
defer cancel()
|
|
if err := s.newapi.verifyKey(ctx, req.Key); err != nil {
|
|
writeError(w, http.StatusUnauthorized, "UNAUTHENTICATED", "key invalid or expired", map[string]any{
|
|
"details": err.Error(),
|
|
})
|
|
return
|
|
}
|
|
|
|
key, u, ok, err := s.store.createSession(req.Key)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to create session", err)
|
|
return
|
|
}
|
|
if !ok {
|
|
writeError(w, http.StatusUnauthorized, "UNAUTHENTICATED", "key invalid or expired", nil)
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"key": key,
|
|
"token": key,
|
|
"user": u,
|
|
})
|
|
}
|
|
|
|
func (s *Server) handleAuthMe(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
current, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, authSessionView{
|
|
User: current.User,
|
|
SessionKey: current.Token,
|
|
UpstreamKeyMasked: maskSecret(current.UpstreamKey),
|
|
})
|
|
}
|
|
|
|
func (s *Server) handleAuthLogout(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
token, ok := bearerToken(r)
|
|
if ok {
|
|
_ = s.store.deleteSession(token)
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{"logged_out": true})
|
|
}
|
|
|
|
func (s *Server) handleAuthTokenRotate(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
token, ok := bearerToken(r)
|
|
if !ok {
|
|
writeError(w, http.StatusUnauthorized, "UNAUTHENTICATED", "missing bearer key", nil)
|
|
return
|
|
}
|
|
newToken, u, exists, err := s.store.rotateSession(token)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to rotate session", err)
|
|
return
|
|
}
|
|
if !exists {
|
|
writeError(w, http.StatusUnauthorized, "UNAUTHENTICATED", "key invalid or expired", nil)
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"key": newToken,
|
|
"token": newToken,
|
|
"user": u,
|
|
})
|
|
}
|
|
|
|
func (s *Server) handleSkills(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
_, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
tag := r.URL.Query().Get("tag")
|
|
search := r.URL.Query().Get("search")
|
|
limit := intQuery(r, "limit", 50)
|
|
offset := intQuery(r, "offset", 0)
|
|
items, total := s.store.listSkills(tag, search, limit, offset)
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"items": items,
|
|
"total": total,
|
|
"limit": limit,
|
|
"offset": offset,
|
|
})
|
|
}
|
|
|
|
func (s *Server) handleSkill(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
_, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
path := strings.TrimPrefix(r.URL.Path, "/v1/skills/")
|
|
if path == "" {
|
|
notFound(w)
|
|
return
|
|
}
|
|
|
|
if strings.HasSuffix(path, "/schema") {
|
|
id := strings.TrimSuffix(strings.TrimSuffix(path, "/schema"), "/")
|
|
item, exists := s.store.getSkill(id)
|
|
if !exists {
|
|
notFound(w)
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"input_schema": item.InputSchema,
|
|
"output_schema": item.OutputSchema,
|
|
})
|
|
return
|
|
}
|
|
|
|
item, exists := s.store.getSkill(strings.Trim(path, "/"))
|
|
if !exists {
|
|
notFound(w)
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, item)
|
|
}
|
|
|
|
func (s *Server) handleJobs(w http.ResponseWriter, r *http.Request) {
|
|
current, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
switch r.Method {
|
|
case http.MethodPost:
|
|
var req struct {
|
|
SkillID string `json:"skill_id"`
|
|
Input map[string]any `json:"input"`
|
|
ProjectID string `json:"project_id"`
|
|
Priority string `json:"priority"`
|
|
IdempotencyKey string `json:"idempotency_key"`
|
|
}
|
|
if !decodeJSON(w, r, &req) {
|
|
return
|
|
}
|
|
if req.SkillID == "" {
|
|
writeError(w, http.StatusBadRequest, "VALIDATION_ERROR", "skill_id is required", nil)
|
|
return
|
|
}
|
|
item, exists := s.store.getSkill(req.SkillID)
|
|
if !exists {
|
|
writeError(w, http.StatusNotFound, "NOT_FOUND", "skill not found", nil)
|
|
return
|
|
}
|
|
routeKey := normalizeRouteKey(item.RouteKey)
|
|
modelID := s.resolveModelID(routeKey, req.ProjectID)
|
|
record, statusCode, err := s.store.createJob(
|
|
req.SkillID,
|
|
routeKey,
|
|
modelID,
|
|
routeExecMode(routeKey),
|
|
req.Input,
|
|
req.ProjectID,
|
|
defaultString(req.Priority, "normal"),
|
|
req.IdempotencyKey,
|
|
current,
|
|
)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to create job", err)
|
|
return
|
|
}
|
|
if statusCode == http.StatusAccepted {
|
|
go s.dispatchJob(record)
|
|
}
|
|
writeData(w, statusCode, map[string]any{
|
|
"job_id": record.JobID,
|
|
"status": record.Status,
|
|
"created_at": record.CreatedAt,
|
|
})
|
|
case http.MethodGet:
|
|
status := r.URL.Query().Get("status")
|
|
skillID := r.URL.Query().Get("skill_id")
|
|
projectID := r.URL.Query().Get("project_id")
|
|
limit := intQuery(r, "limit", 20)
|
|
offset := intQuery(r, "offset", 0)
|
|
items, total, err := s.store.listJobs(current.User.ID, status, skillID, projectID, limit, offset)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to list jobs", err)
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"items": items,
|
|
"total": total,
|
|
"limit": limit,
|
|
"offset": offset,
|
|
})
|
|
default:
|
|
methodNotAllowed(w)
|
|
}
|
|
}
|
|
|
|
func (s *Server) handleJob(w http.ResponseWriter, r *http.Request) {
|
|
current, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
path := strings.TrimPrefix(r.URL.Path, "/v1/jobs/")
|
|
parts := strings.Split(strings.Trim(path, "/"), "/")
|
|
if len(parts) == 0 || parts[0] == "" {
|
|
notFound(w)
|
|
return
|
|
}
|
|
jobID := parts[0]
|
|
|
|
if len(parts) == 1 && r.Method == http.MethodGet {
|
|
record, exists, err := s.store.getJob(current.User.ID, jobID)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to load job", err)
|
|
return
|
|
}
|
|
if !exists {
|
|
notFound(w)
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, record)
|
|
return
|
|
}
|
|
|
|
if len(parts) == 2 && parts[1] == "cancel" && r.Method == http.MethodPost {
|
|
record, exists, conflict, err := s.store.cancelJob(current.User.ID, jobID)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to cancel job", err)
|
|
return
|
|
}
|
|
if !exists {
|
|
notFound(w)
|
|
return
|
|
}
|
|
if conflict {
|
|
writeError(w, http.StatusConflict, "CONFLICT", "job already in terminal state", nil)
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, record)
|
|
return
|
|
}
|
|
|
|
if len(parts) == 2 && parts[1] == "logs" && r.Method == http.MethodGet {
|
|
logs, exists, err := s.store.jobLogs(current.User.ID, jobID)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to load job logs", err)
|
|
return
|
|
}
|
|
if !exists {
|
|
notFound(w)
|
|
return
|
|
}
|
|
if strings.Contains(r.Header.Get("Accept"), "text/event-stream") {
|
|
w.Header().Set("Content-Type", "text/event-stream")
|
|
w.Header().Set("Cache-Control", "no-cache")
|
|
flusher, _ := w.(http.Flusher)
|
|
for _, item := range logs {
|
|
fmt.Fprintf(w, "data: %s\n\n", mustJSON(item))
|
|
if flusher != nil {
|
|
flusher.Flush()
|
|
}
|
|
time.Sleep(150 * time.Millisecond)
|
|
}
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, logs)
|
|
return
|
|
}
|
|
|
|
if len(parts) == 2 && parts[1] == "artifacts" && r.Method == http.MethodGet {
|
|
items, exists, err := s.store.artifactsForJob(current.User.ID, jobID)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to list job artifacts", err)
|
|
return
|
|
}
|
|
if !exists {
|
|
notFound(w)
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{"items": items})
|
|
return
|
|
}
|
|
|
|
notFound(w)
|
|
}
|
|
|
|
func (s *Server) handleArtifactUpload(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
current, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
r.Body = http.MaxBytesReader(w, r.Body, 20<<20)
|
|
if err := r.ParseMultipartForm(20 << 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
|
|
}
|
|
|
|
var metadata any
|
|
metadataJSON := strings.TrimSpace(r.FormValue("metadata_json"))
|
|
if metadataJSON != "" {
|
|
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 = "artifact" + extensionFromContentType(contentType)
|
|
}
|
|
filename = sanitizeFilename(filename)
|
|
visibility := strings.TrimSpace(r.FormValue("visibility"))
|
|
|
|
input := map[string]any{
|
|
"filename": filename,
|
|
"content_type": contentType,
|
|
"size_bytes": len(content),
|
|
"role": strings.TrimSpace(r.FormValue("role")),
|
|
}
|
|
projectID := strings.TrimSpace(r.FormValue("project_id"))
|
|
if projectID != "" {
|
|
input["project_id"] = projectID
|
|
}
|
|
if metadata != nil {
|
|
input["metadata"] = metadata
|
|
}
|
|
|
|
record, _, err := s.store.createJob(
|
|
"popiskill-artifact-upload-local-v1",
|
|
"artifact.upload",
|
|
"",
|
|
"local_upload",
|
|
input,
|
|
projectID,
|
|
"normal",
|
|
"",
|
|
current,
|
|
)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to create upload job", err)
|
|
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: "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)
|
|
return
|
|
}
|
|
|
|
item, _, exists, err := s.store.artifactRef(current.User.ID, buildArtifactID(record.JobID, 0))
|
|
if err != nil {
|
|
writeInternalError(w, "failed to load uploaded artifact", err)
|
|
return
|
|
}
|
|
if !exists || item == nil {
|
|
writeInternalError(w, "uploaded artifact missing after persistence", errors.New("artifact not found"))
|
|
return
|
|
}
|
|
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 {
|
|
return
|
|
}
|
|
|
|
path := strings.TrimPrefix(r.URL.Path, "/v1/artifacts/")
|
|
parts := strings.Split(strings.Trim(path, "/"), "/")
|
|
if len(parts) == 0 || parts[0] == "" {
|
|
notFound(w)
|
|
return
|
|
}
|
|
artifactID := parts[0]
|
|
item, ref, exists, err := s.store.artifactRef(current.User.ID, artifactID)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to load artifact", err)
|
|
return
|
|
}
|
|
if !exists {
|
|
notFound(w)
|
|
return
|
|
}
|
|
|
|
if len(parts) == 1 && r.Method == http.MethodGet {
|
|
writeData(w, http.StatusOK, item)
|
|
return
|
|
}
|
|
|
|
if len(parts) == 2 && parts[1] == "content" && r.Method == http.MethodGet {
|
|
ctx, cancel := context.WithTimeout(r.Context(), 3*time.Minute)
|
|
defer cancel()
|
|
contentType, sizeBytes, reader, err := s.newapi.openResultRef(ctx, current.UpstreamKey, ref)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to read artifact content", err)
|
|
return
|
|
}
|
|
defer reader.Close()
|
|
w.Header().Set("Content-Type", defaultString(contentType, item.ContentType))
|
|
if sizeBytes > 0 {
|
|
w.Header().Set("Content-Length", strconv.FormatInt(sizeBytes, 10))
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
if _, err := io.Copy(w, reader); err != nil {
|
|
log.Printf("popiartServer: streaming artifact %s failed: %v", artifactID, err)
|
|
}
|
|
return
|
|
}
|
|
|
|
notFound(w)
|
|
}
|
|
|
|
func (s *Server) handleBudget(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
_, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"period": map[string]any{
|
|
"start": time.Now().UTC().AddDate(0, 0, -30).Format(time.RFC3339),
|
|
"end": time.Now().UTC().Format(time.RFC3339),
|
|
},
|
|
"used": map[string]any{
|
|
"tokens": 12000,
|
|
"cost_usd": 4.62,
|
|
},
|
|
"limit": map[string]any{
|
|
"monthly_tokens": 1000000,
|
|
"monthly_cost_usd": 200,
|
|
},
|
|
"remaining": map[string]any{
|
|
"tokens": 988000,
|
|
"cost_usd": 195.38,
|
|
},
|
|
})
|
|
}
|
|
|
|
func (s *Server) handleBudgetUsage(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
_, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"rows": []map[string]any{
|
|
{"dimension": "popiskill-image-text2image-basic-v1", "tokens_used": 3000, "cost_usd": 1.2, "job_count": 3},
|
|
{"dimension": "popiskill-video-image2video-basic-v1", "tokens_used": 9000, "cost_usd": 3.42, "job_count": 2},
|
|
},
|
|
"total": map[string]any{"tokens_used": 12000, "cost_usd": 4.62, "job_count": 5},
|
|
})
|
|
}
|
|
|
|
func (s *Server) handleBudgetLimits(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
_, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"rate_limits": []map[string]any{
|
|
{"window": "1m", "limit": 60, "remaining": 59, "reset_at": time.Now().UTC().Add(time.Minute).Format(time.RFC3339)},
|
|
},
|
|
"quota": map[string]any{"monthly_tokens": 1000000, "monthly_cost_usd": 200},
|
|
})
|
|
}
|
|
|
|
func (s *Server) handleProjects(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
_, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"items": s.store.projects,
|
|
"total": len(s.store.projects),
|
|
"limit": intQuery(r, "limit", 20),
|
|
"offset": 0,
|
|
})
|
|
}
|
|
|
|
func (s *Server) handleProject(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
_, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
path := strings.TrimPrefix(r.URL.Path, "/v1/projects/")
|
|
parts := strings.Split(strings.Trim(path, "/"), "/")
|
|
if len(parts) == 0 || parts[0] == "" {
|
|
notFound(w)
|
|
return
|
|
}
|
|
projectID := parts[0]
|
|
|
|
var found *project
|
|
for _, item := range s.store.projects {
|
|
if item.ID == projectID {
|
|
copyItem := item
|
|
found = ©Item
|
|
break
|
|
}
|
|
}
|
|
if found == nil {
|
|
notFound(w)
|
|
return
|
|
}
|
|
|
|
if len(parts) == 1 {
|
|
writeData(w, http.StatusOK, found)
|
|
return
|
|
}
|
|
|
|
if len(parts) == 2 && parts[1] == "context" {
|
|
routes, err := s.store.routesForProject(projectID)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to load project routes", err)
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"project": found,
|
|
"budget": map[string]any{
|
|
"used": map[string]any{"tokens": 12000, "cost_usd": 4.62},
|
|
"limit": map[string]any{"monthly_tokens": 1000000, "monthly_cost_usd": 200},
|
|
"remaining": map[string]any{"tokens": 988000, "cost_usd": 195.38},
|
|
},
|
|
"route_overrides": routes,
|
|
"available_skills": s.store.skills,
|
|
})
|
|
return
|
|
}
|
|
|
|
notFound(w)
|
|
}
|
|
|
|
func (s *Server) handleModels(w http.ResponseWriter, r *http.Request) {
|
|
if r.URL.Path != "/v1/models" {
|
|
notFound(w)
|
|
return
|
|
}
|
|
current, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
if r.Method != http.MethodGet {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
if s.newapi == nil || !s.newapi.enabled() {
|
|
writeError(w, http.StatusServiceUnavailable, "MODEL_LIST_UNAVAILABLE", "PopiNewAPI is not configured for popiartServer", nil)
|
|
return
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
|
|
defer cancel()
|
|
items, err := s.newapi.listModels(ctx, current.UpstreamKey)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadGateway, "MODEL_LIST_FAILED", "failed to load models from PopiNewAPI", map[string]any{
|
|
"details": err.Error(),
|
|
})
|
|
return
|
|
}
|
|
|
|
modelType := r.URL.Query().Get("type")
|
|
provider := r.URL.Query().Get("provider")
|
|
filtered := make([]model, 0, len(items))
|
|
for _, item := range items {
|
|
if modelType != "" && item.Type != modelType {
|
|
continue
|
|
}
|
|
if provider != "" && item.Provider != provider {
|
|
continue
|
|
}
|
|
filtered = append(filtered, item)
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{"items": filtered})
|
|
}
|
|
|
|
func (s *Server) handleModelRoutes(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
_, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
projectID := r.URL.Query().Get("project_id")
|
|
routes, err := s.store.routesForProject(projectID)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to load route overrides", err)
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"global": map[string]any{
|
|
"image": s.cfg.DefaultImageModel,
|
|
"video": s.cfg.DefaultVideoModel,
|
|
"audio": "",
|
|
},
|
|
"overrides": map[string]any{
|
|
projectID: routes,
|
|
},
|
|
})
|
|
}
|
|
|
|
func (s *Server) handleModelRouteOverrides(w http.ResponseWriter, r *http.Request) {
|
|
_, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
switch r.Method {
|
|
case http.MethodGet:
|
|
projectID := r.URL.Query().Get("project_id")
|
|
routes, err := s.store.routesForProject(projectID)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to load route overrides", err)
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"project_id": projectID,
|
|
"items": routes,
|
|
})
|
|
case http.MethodPost:
|
|
var req struct {
|
|
ProjectID string `json:"project_id"`
|
|
SkillType string `json:"skill_type"`
|
|
ModelID string `json:"model_id"`
|
|
}
|
|
if !decodeJSON(w, r, &req) {
|
|
return
|
|
}
|
|
if req.ProjectID == "" || req.SkillType == "" || req.ModelID == "" {
|
|
writeError(w, http.StatusBadRequest, "VALIDATION_ERROR", "project_id, skill_type, and model_id are required", nil)
|
|
return
|
|
}
|
|
routeKey := normalizeRouteKey(req.SkillType)
|
|
if err := s.store.setRouteOverride(req.ProjectID, routeKey, req.ModelID); err != nil {
|
|
writeInternalError(w, "failed to save route override", err)
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"project_id": req.ProjectID,
|
|
"skill_type": req.SkillType,
|
|
"route_key": routeKey,
|
|
"model_id": req.ModelID,
|
|
})
|
|
default:
|
|
methodNotAllowed(w)
|
|
}
|
|
}
|
|
|
|
func (s *Server) handleModelRouteOverrideUnset(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
_, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
var req struct {
|
|
ProjectID string `json:"project_id"`
|
|
SkillType string `json:"skill_type"`
|
|
}
|
|
if !decodeJSON(w, r, &req) {
|
|
return
|
|
}
|
|
if req.ProjectID == "" || req.SkillType == "" {
|
|
writeError(w, http.StatusBadRequest, "VALIDATION_ERROR", "project_id and skill_type are required", nil)
|
|
return
|
|
}
|
|
routeKey := normalizeRouteKey(req.SkillType)
|
|
if err := s.store.unsetRouteOverride(req.ProjectID, routeKey); err != nil {
|
|
writeInternalError(w, "failed to delete route override", err)
|
|
return
|
|
}
|
|
writeData(w, http.StatusOK, map[string]any{
|
|
"project_id": req.ProjectID,
|
|
"skill_type": req.SkillType,
|
|
"route_key": routeKey,
|
|
"unset": true,
|
|
})
|
|
}
|
|
|
|
func (s *Server) handleModelsInfer(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
methodNotAllowed(w)
|
|
return
|
|
}
|
|
current, ok := s.authenticateSession(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
var req struct {
|
|
ModelID string `json:"model_id"`
|
|
Input map[string]any `json:"input"`
|
|
ProjectID string `json:"project_id"`
|
|
Priority string `json:"priority"`
|
|
IdempotencyKey string `json:"idempotency_key"`
|
|
}
|
|
if !decodeJSON(w, r, &req) {
|
|
return
|
|
}
|
|
if req.ModelID == "" {
|
|
writeError(w, http.StatusBadRequest, "VALIDATION_ERROR", "model_id is required", nil)
|
|
return
|
|
}
|
|
routeKey := inferRouteKeyForModel(req.ModelID)
|
|
record, statusCode, err := s.store.createJob(
|
|
"",
|
|
routeKey,
|
|
req.ModelID,
|
|
routeExecMode(routeKey),
|
|
req.Input,
|
|
req.ProjectID,
|
|
defaultString(req.Priority, "normal"),
|
|
req.IdempotencyKey,
|
|
current,
|
|
)
|
|
if err != nil {
|
|
writeInternalError(w, "failed to create model job", err)
|
|
return
|
|
}
|
|
if statusCode == http.StatusAccepted {
|
|
go s.dispatchJob(record)
|
|
}
|
|
writeData(w, statusCode, map[string]any{
|
|
"job_id": record.JobID,
|
|
"status": record.Status,
|
|
"created_at": record.CreatedAt,
|
|
})
|
|
}
|
|
|
|
func (s *Server) dispatchJob(record *job) {
|
|
if record == nil {
|
|
return
|
|
}
|
|
|
|
switch normalizeRouteKey(record.RouteKey) {
|
|
case "image.text2image":
|
|
s.executeTextToImageJob(record)
|
|
case "image.img2img":
|
|
s.executeImageToImageJob(record)
|
|
case "video.image2video":
|
|
s.executeImageToVideoJob(record)
|
|
default:
|
|
s.executeUnsupportedSkill(record)
|
|
}
|
|
}
|
|
|
|
func (s *Server) executeTextToImageJob(record *job) {
|
|
if _, _, err := s.store.startJob(record.JobID); err != nil {
|
|
log.Printf("popiartServer: start job %s failed: %v", record.JobID, err)
|
|
}
|
|
|
|
if s.newapi == nil || !s.newapi.enabled() {
|
|
if err := s.store.failJob(record.JobID, "NEWAPI_NOT_CONFIGURED", "PopiNewAPI is not configured for popiartServer", nil); err != nil {
|
|
log.Printf("popiartServer: fail job %s failed: %v", record.JobID, err)
|
|
}
|
|
return
|
|
}
|
|
|
|
modelID := strings.TrimSpace(record.ModelID)
|
|
if modelID == "" {
|
|
modelID = s.resolveModelID(record.RouteKey, record.ProjectID)
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
|
|
defer cancel()
|
|
|
|
refs, usage, err := s.newapi.generateImageRefs(ctx, record.UpstreamKey, modelID, record.Input)
|
|
if err != nil {
|
|
if repoErr := s.store.failJob(record.JobID, "MODEL_REQUEST_FAILED", err.Error(), map[string]any{
|
|
"model_id": modelID,
|
|
}); repoErr != nil {
|
|
log.Printf("popiartServer: fail job %s failed: %v", record.JobID, repoErr)
|
|
}
|
|
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{
|
|
"details": err.Error(),
|
|
}); failErr != nil {
|
|
log.Printf("popiartServer: fail job %s after result ref error failed: %v", record.JobID, failErr)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Server) executeImageToImageJob(record *job) {
|
|
if _, _, err := s.store.startJob(record.JobID); err != nil {
|
|
log.Printf("popiartServer: start job %s failed: %v", record.JobID, err)
|
|
}
|
|
|
|
if s.newapi == nil || !s.newapi.enabled() {
|
|
if err := s.store.failJob(record.JobID, "NEWAPI_NOT_CONFIGURED", "PopiNewAPI is not configured for popiartServer", nil); err != nil {
|
|
log.Printf("popiartServer: fail job %s failed: %v", record.JobID, err)
|
|
}
|
|
return
|
|
}
|
|
|
|
modelID := strings.TrimSpace(record.ModelID)
|
|
if modelID == "" {
|
|
modelID = s.resolveModelID(record.RouteKey, record.ProjectID)
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
|
|
defer cancel()
|
|
|
|
input := buildImageToImageInput(record)
|
|
ref, err := s.resolveImageToImageReference(ctx, record, input)
|
|
if err != nil {
|
|
if repoErr := s.store.failJob(record.JobID, "REFERENCE_IMAGE_RESOLUTION_FAILED", err.Error(), map[string]any{
|
|
"skill_id": record.SkillID,
|
|
"route_key": record.RouteKey,
|
|
}); repoErr != nil {
|
|
log.Printf("popiartServer: fail job %s failed: %v", record.JobID, repoErr)
|
|
}
|
|
return
|
|
}
|
|
|
|
refs, usage, err := s.newapi.generateEditedImageRefs(ctx, record.UpstreamKey, modelID, input, ref)
|
|
if err != nil {
|
|
if repoErr := s.store.failJob(record.JobID, "MODEL_REQUEST_FAILED", err.Error(), map[string]any{
|
|
"model_id": modelID,
|
|
"route_key": record.RouteKey,
|
|
}); repoErr != nil {
|
|
log.Printf("popiartServer: fail job %s failed: %v", record.JobID, repoErr)
|
|
}
|
|
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{
|
|
"details": err.Error(),
|
|
}); failErr != nil {
|
|
log.Printf("popiartServer: fail job %s after result ref error failed: %v", record.JobID, failErr)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Server) executeImageToVideoJob(record *job) {
|
|
if _, _, err := s.store.startJob(record.JobID); err != nil {
|
|
log.Printf("popiartServer: start job %s failed: %v", record.JobID, err)
|
|
}
|
|
|
|
if s.newapi == nil || !s.newapi.enabled() {
|
|
if err := s.store.failJob(record.JobID, "NEWAPI_NOT_CONFIGURED", "PopiNewAPI is not configured for popiartServer", nil); err != nil {
|
|
log.Printf("popiartServer: fail job %s failed: %v", record.JobID, err)
|
|
}
|
|
return
|
|
}
|
|
|
|
modelID := strings.TrimSpace(record.ModelID)
|
|
if modelID == "" {
|
|
modelID = s.resolveModelID(record.RouteKey, record.ProjectID)
|
|
}
|
|
|
|
input := buildImageToVideoInput(record)
|
|
submitCtx, cancelSubmit := context.WithTimeout(context.Background(), 2*time.Minute)
|
|
defer cancelSubmit()
|
|
|
|
ref, err := s.resolveImageToImageReference(submitCtx, record, input)
|
|
if err != nil {
|
|
if repoErr := s.store.failJob(record.JobID, "REFERENCE_IMAGE_RESOLUTION_FAILED", err.Error(), map[string]any{
|
|
"skill_id": record.SkillID,
|
|
"route_key": record.RouteKey,
|
|
}); repoErr != nil {
|
|
log.Printf("popiartServer: fail job %s failed: %v", record.JobID, repoErr)
|
|
}
|
|
return
|
|
}
|
|
|
|
upstreamTaskID, err := s.newapi.submitImageToVideoTask(submitCtx, record.UpstreamKey, modelID, input, ref)
|
|
if err != nil {
|
|
if repoErr := s.store.failJob(record.JobID, "MODEL_REQUEST_FAILED", err.Error(), map[string]any{
|
|
"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.linkUpstreamTask(record.JobID, upstreamTaskID); err != nil {
|
|
if repoErr := s.store.failJob(record.JobID, "TASK_LINK_PERSIST_FAILED", "failed to persist upstream task id", map[string]any{
|
|
"details": err.Error(),
|
|
"newapi_task_id": upstreamTaskID,
|
|
}); repoErr != nil {
|
|
log.Printf("popiartServer: fail job %s failed: %v", record.JobID, repoErr)
|
|
}
|
|
return
|
|
}
|
|
|
|
pollCtx, cancelPoll := context.WithTimeout(context.Background(), 20*time.Minute)
|
|
defer cancelPoll()
|
|
|
|
taskResult, err := s.waitForVideoTask(pollCtx, record.UpstreamKey, upstreamTaskID)
|
|
if err != nil {
|
|
if repoErr := s.store.failJob(record.JobID, "TASK_POLL_FAILED", err.Error(), map[string]any{
|
|
"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
|
|
}
|
|
|
|
refs := []resultRef{buildVideoResultRef(s.newapi.baseURL, upstreamTaskID, modelID, taskResult)}
|
|
usage := map[string]any{
|
|
"newapi_task_id": upstreamTaskID,
|
|
"status": taskResult.Status,
|
|
}
|
|
if strings.TrimSpace(taskResult.Format) != "" {
|
|
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{
|
|
"details": err.Error(),
|
|
}); failErr != nil {
|
|
log.Printf("popiartServer: fail job %s after result ref error failed: %v", record.JobID, failErr)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Server) executeUnsupportedSkill(record *job) {
|
|
if _, _, err := s.store.startJob(record.JobID); err != nil {
|
|
log.Printf("popiartServer: start job %s failed: %v", record.JobID, err)
|
|
}
|
|
time.Sleep(300 * time.Millisecond)
|
|
if err := s.store.failJob(record.JobID, "NOT_IMPLEMENTED", "skill runtime is not connected yet", map[string]any{
|
|
"skill_id": record.SkillID,
|
|
"route_key": record.RouteKey,
|
|
"exec_mode": record.ExecMode,
|
|
}); err != nil {
|
|
log.Printf("popiartServer: fail job %s failed: %v", record.JobID, err)
|
|
}
|
|
}
|
|
|
|
func (s *Server) resolveModelID(routeKey, projectID string) string {
|
|
routes, err := s.store.routesForProject(projectID)
|
|
if err == nil {
|
|
if override := routes[normalizeRouteKey(routeKey)]; override != "" {
|
|
return override
|
|
}
|
|
}
|
|
switch normalizeRouteKey(routeKey) {
|
|
case "video.image2video":
|
|
return s.cfg.DefaultVideoModel
|
|
default:
|
|
return s.cfg.DefaultImageModel
|
|
}
|
|
}
|
|
|
|
func inferRouteKeyForModel(modelID string) string {
|
|
if modelType, _ := classifyModelIDFallback(modelID); modelType == "video" {
|
|
return "video.image2video"
|
|
}
|
|
return "image.text2image"
|
|
}
|
|
|
|
func (s *Server) resolveImageToImageReference(ctx context.Context, record *job, input map[string]any) (imageEditReference, error) {
|
|
if record == nil {
|
|
return imageEditReference{}, fmt.Errorf("job record is required")
|
|
}
|
|
if artifactID := strings.TrimSpace(stringValue(input["source_artifact_id"])); artifactID != "" {
|
|
item, ref, exists, err := s.store.artifactRef(record.UserID, artifactID)
|
|
if err != nil {
|
|
return imageEditReference{}, err
|
|
}
|
|
if !exists {
|
|
return imageEditReference{}, fmt.Errorf("source artifact not found: %s", artifactID)
|
|
}
|
|
contentType, _, reader, err := s.newapi.openResultRef(ctx, record.UpstreamKey, ref)
|
|
if err != nil {
|
|
return imageEditReference{}, err
|
|
}
|
|
defer reader.Close()
|
|
content, err := io.ReadAll(reader)
|
|
if err != nil {
|
|
return imageEditReference{}, err
|
|
}
|
|
return imageEditReference{
|
|
Filename: item.Filename,
|
|
ContentType: defaultString(contentType, item.ContentType),
|
|
Content: content,
|
|
URL: item.URL,
|
|
}, nil
|
|
}
|
|
|
|
refURL := strings.TrimSpace(stringValue(
|
|
input["reference_image_url"],
|
|
input["image_url"],
|
|
))
|
|
if refURL == "" && strings.Contains(record.SkillID, "popistudio-alice-showcase") {
|
|
refURL = defaultAliceReferenceURL
|
|
}
|
|
if refURL == "" {
|
|
return imageEditReference{}, fmt.Errorf("reference image is required")
|
|
}
|
|
return s.downloadReferenceImage(ctx, refURL)
|
|
}
|
|
|
|
func (s *Server) waitForVideoTask(ctx context.Context, token, taskID string) (*videoTaskResult, error) {
|
|
ticker := time.NewTicker(3 * time.Second)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
result, err := s.newapi.fetchVideoTask(ctx, token, taskID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
switch result.Status {
|
|
case "completed":
|
|
return result, nil
|
|
case "failed":
|
|
message := defaultString(strings.TrimSpace(result.ErrorReason), "video task failed")
|
|
return nil, errors.New(message)
|
|
case "queued", "in_progress", "":
|
|
default:
|
|
if strings.TrimSpace(result.Status) != "" {
|
|
return nil, fmt.Errorf("unexpected video task status: %s", result.Status)
|
|
}
|
|
}
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Server) downloadReferenceImage(ctx context.Context, rawURL string) (imageEditReference, error) {
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil)
|
|
if err != nil {
|
|
return imageEditReference{}, err
|
|
}
|
|
resp, err := s.newapi.httpClient.Do(req)
|
|
if err != nil {
|
|
return imageEditReference{}, err
|
|
}
|
|
defer resp.Body.Close()
|
|
if resp.StatusCode >= 400 {
|
|
body, _ := io.ReadAll(resp.Body)
|
|
return imageEditReference{}, fmt.Errorf("download reference image failed: %s", strings.TrimSpace(string(body)))
|
|
}
|
|
content, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return imageEditReference{}, err
|
|
}
|
|
contentType := defaultString(strings.TrimSpace(resp.Header.Get("Content-Type")), http.DetectContentType(content))
|
|
return imageEditReference{
|
|
Filename: filenameFromURL(rawURL, "reference"+extensionFromContentType(contentType)),
|
|
ContentType: contentType,
|
|
Content: content,
|
|
URL: strings.TrimSpace(rawURL),
|
|
}, nil
|
|
}
|
|
|
|
func buildImageToImageInput(record *job) map[string]any {
|
|
input := cloneMap(record.Input)
|
|
if input == nil {
|
|
input = map[string]any{}
|
|
}
|
|
|
|
prompt := strings.TrimSpace(stringValue(
|
|
input["prompt"],
|
|
input["scene_prompt"],
|
|
))
|
|
|
|
if strings.Contains(record.SkillID, "popistudio-alice-showcase") {
|
|
prompt = buildAliceShowcasePrompt(input, prompt)
|
|
} else {
|
|
prompt = buildGenericImageToImagePrompt(input, prompt)
|
|
}
|
|
|
|
input["prompt"] = prompt
|
|
delete(input, "scene_prompt")
|
|
return input
|
|
}
|
|
|
|
func buildImageToVideoInput(record *job) map[string]any {
|
|
input := cloneMap(record.Input)
|
|
if input == nil {
|
|
input = map[string]any{}
|
|
}
|
|
|
|
prompt := strings.TrimSpace(stringValue(
|
|
input["prompt"],
|
|
input["motion_prompt"],
|
|
input["scene_prompt"],
|
|
))
|
|
|
|
parts := make([]string, 0, 6)
|
|
if prompt != "" {
|
|
parts = append(parts, prompt)
|
|
}
|
|
if cameraMotion := strings.TrimSpace(stringValue(input["camera_motion"])); cameraMotion != "" {
|
|
parts = append(parts, "camera motion "+cameraMotion)
|
|
}
|
|
if shotType := strings.TrimSpace(stringValue(input["shot_type"])); shotType != "" {
|
|
parts = append(parts, shotType)
|
|
}
|
|
if mood := strings.TrimSpace(stringValue(input["mood"])); mood != "" {
|
|
parts = append(parts, mood)
|
|
}
|
|
if aspectRatio := strings.TrimSpace(stringValue(input["aspect_ratio"])); aspectRatio != "" {
|
|
parts = append(parts, "aspect ratio "+aspectRatio)
|
|
}
|
|
if len(parts) == 0 {
|
|
parts = append(parts, "Generate a short polished cinematic motion clip that preserves the reference image subject identity.")
|
|
}
|
|
|
|
input["prompt"] = strings.Join(parts, ", ")
|
|
delete(input, "motion_prompt")
|
|
delete(input, "scene_prompt")
|
|
return input
|
|
}
|
|
|
|
func buildAliceShowcasePrompt(input map[string]any, scenePrompt string) string {
|
|
parts := []string{
|
|
"PopiStudio Alice as the same fixed main protagonist from the canonical Alice reference image",
|
|
"preserve the same anime facial structure, recognizability, hairstyle, hair color, clothing language, and main palette",
|
|
"Alice remains the visual center of the frame",
|
|
}
|
|
|
|
if scenePrompt != "" {
|
|
parts = append(parts, scenePrompt)
|
|
}
|
|
if shotType := strings.TrimSpace(stringValue(input["shot_type"])); shotType != "" {
|
|
parts = append(parts, shotType)
|
|
}
|
|
if camera := strings.TrimSpace(stringValue(input["camera"])); camera != "" {
|
|
parts = append(parts, camera)
|
|
}
|
|
if mood := strings.TrimSpace(stringValue(input["mood"])); mood != "" {
|
|
parts = append(parts, mood)
|
|
}
|
|
if aspectRatio := strings.TrimSpace(stringValue(input["aspect_ratio"])); aspectRatio != "" {
|
|
parts = append(parts, "aspect ratio "+aspectRatio)
|
|
}
|
|
|
|
parts = append(parts,
|
|
"real-world modern Chinese living scene",
|
|
"Sony photo realism",
|
|
"natural light",
|
|
"warm realistic tone",
|
|
"high detail",
|
|
"clean composition",
|
|
)
|
|
|
|
return strings.Join(parts, ", ")
|
|
}
|
|
|
|
func buildGenericImageToImagePrompt(input map[string]any, prompt string) string {
|
|
parts := make([]string, 0, 6)
|
|
if prompt != "" {
|
|
parts = append(parts, prompt)
|
|
}
|
|
if refURL := strings.TrimSpace(stringValue(input["reference_image_url"], input["image_url"])); refURL != "" {
|
|
parts = append(parts, "preserve the same main subject identity and key visual traits from the reference image")
|
|
}
|
|
if aspectRatio := strings.TrimSpace(stringValue(input["aspect_ratio"])); aspectRatio != "" {
|
|
parts = append(parts, "aspect ratio "+aspectRatio)
|
|
}
|
|
if len(parts) == 0 {
|
|
return "preserve the same main subject identity and render a polished high-detail image"
|
|
}
|
|
return strings.Join(parts, ", ")
|
|
}
|
|
|
|
func stringValue(values ...any) string {
|
|
for _, value := range values {
|
|
if text, ok := value.(string); ok && strings.TrimSpace(text) != "" {
|
|
return text
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func buildVideoResultRef(baseURL, taskID, modelID string, task *videoTaskResult) resultRef {
|
|
urlValue := strings.TrimSpace(task.URL)
|
|
if urlValue == "" {
|
|
urlValue = strings.TrimRight(baseURL, "/") + "/v1/videos/" + taskID + "/content"
|
|
}
|
|
|
|
contentType := "video/mp4"
|
|
ext := ".mp4"
|
|
if format := strings.TrimSpace(task.Format); format != "" {
|
|
switch strings.ToLower(strings.TrimPrefix(format, ".")) {
|
|
case "mov", "quicktime":
|
|
contentType = "video/quicktime"
|
|
ext = ".mov"
|
|
case "webm":
|
|
contentType = "video/webm"
|
|
ext = ".webm"
|
|
case "mp4":
|
|
contentType = "video/mp4"
|
|
ext = ".mp4"
|
|
}
|
|
}
|
|
filename := filenameFromURL(urlValue, "")
|
|
if filename == "" || filename == "content" || !strings.Contains(filepath.Base(filename), ".") {
|
|
filename = sanitizeFilename(defaultString(modelID, "video")) + "-" + sanitizeFilename(taskID) + ext
|
|
}
|
|
|
|
if strings.HasPrefix(urlValue, "data:") {
|
|
return resultRef{
|
|
Kind: "data_url",
|
|
DataURL: urlValue,
|
|
Filename: filename,
|
|
ContentType: contentType,
|
|
}
|
|
}
|
|
return resultRef{
|
|
Kind: "url",
|
|
URL: urlValue,
|
|
Filename: filename,
|
|
ContentType: contentType,
|
|
}
|
|
}
|
|
|
|
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)
|
|
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)
|
|
return session{}, false
|
|
}
|
|
return current, true
|
|
}
|
|
|
|
func bearerToken(r *http.Request) (string, bool) {
|
|
header := strings.TrimSpace(r.Header.Get("Authorization"))
|
|
if header == "" || !strings.HasPrefix(header, "Bearer ") {
|
|
return "", false
|
|
}
|
|
return strings.TrimSpace(strings.TrimPrefix(header, "Bearer ")), true
|
|
}
|
|
|
|
func decodeJSON(w http.ResponseWriter, r *http.Request, dst any) bool {
|
|
defer r.Body.Close()
|
|
body, err := io.ReadAll(r.Body)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "BAD_REQUEST", "failed to read request body", nil)
|
|
return false
|
|
}
|
|
if len(strings.TrimSpace(string(body))) == 0 {
|
|
body = []byte(`{}`)
|
|
}
|
|
if err := json.Unmarshal(body, dst); err != nil {
|
|
writeError(w, http.StatusBadRequest, "VALIDATION_ERROR", "invalid JSON body", map[string]any{
|
|
"details": err.Error(),
|
|
})
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
func intQuery(r *http.Request, key string, fallback int) int {
|
|
value := r.URL.Query().Get(key)
|
|
if value == "" {
|
|
return fallback
|
|
}
|
|
n, err := strconv.Atoi(value)
|
|
if err != nil || n < 0 {
|
|
return fallback
|
|
}
|
|
return n
|
|
}
|
|
|
|
func writeData(w http.ResponseWriter, status int, data any) {
|
|
writeJSON(w, status, map[string]any{
|
|
"ok": true,
|
|
"data": data,
|
|
})
|
|
}
|
|
|
|
func buildDataURL(contentType string, content []byte) string {
|
|
return "data:" + defaultString(strings.TrimSpace(contentType), "application/octet-stream") + ";base64," + base64.StdEncoding.EncodeToString(content)
|
|
}
|
|
|
|
func writeError(w http.ResponseWriter, status int, code, message string, details map[string]any) {
|
|
errBody := map[string]any{
|
|
"code": code,
|
|
"message": message,
|
|
}
|
|
for key, value := range details {
|
|
errBody[key] = value
|
|
}
|
|
writeJSON(w, status, map[string]any{
|
|
"ok": false,
|
|
"error": errBody,
|
|
})
|
|
}
|
|
|
|
func writeInternalError(w http.ResponseWriter, message string, err error) {
|
|
details := map[string]any{}
|
|
if err != nil {
|
|
details["details"] = err.Error()
|
|
}
|
|
writeError(w, http.StatusInternalServerError, "SERVER_ERROR", message, details)
|
|
}
|
|
|
|
func writeJSON(w http.ResponseWriter, status int, payload any) {
|
|
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
|
w.WriteHeader(status)
|
|
_ = json.NewEncoder(w).Encode(payload)
|
|
}
|
|
|
|
func notFound(w http.ResponseWriter) {
|
|
writeError(w, http.StatusNotFound, "NOT_FOUND", "resource not found", nil)
|
|
}
|
|
|
|
func methodNotAllowed(w http.ResponseWriter) {
|
|
writeError(w, http.StatusMethodNotAllowed, "BAD_REQUEST", "method not allowed", nil)
|
|
}
|
|
|
|
func mustJSON(value any) string {
|
|
data, err := json.Marshal(value)
|
|
if err != nil {
|
|
return `{"level":"error","message":"marshal failed"}`
|
|
}
|
|
return string(data)
|
|
}
|
|
|
|
func defaultString(value, fallback string) string {
|
|
if value != "" {
|
|
return value
|
|
}
|
|
return fallback
|
|
}
|
|
|