channels.
+ // Without a real user identity, events go to user:anonymous and never reach the client.
+ r.Group(func(r app.Router) {
+ r.Use(auth.Middleware(auth.MiddlewareConfig{
+ Validator: jwtValidator,
+ }))
+ r.Use(auth.SessionCheck(sessionChecker))
+
+ // Chat messaging
+ r.Post("/chat/messages", app.Wrap(chatHandler.SendMessage))
+
+ // Media generation (all queue-based, returns 202)
+ r.Post("/generate/image", app.Wrap(generateHandler.GenerateImage))
+ r.Post("/generate/video", app.Wrap(generateHandler.GenerateVideo))
+ r.Post("/generate/text", app.Wrap(generateHandler.GenerateText))
+ r.Get("/generate/jobs/{id}", app.Wrap(generateHandler.GetJobStatus))
+
+ // Media library (upload, list, delete)
+ r.Mount("/media", mediaHandler.Routes())
+
+ // Album generation (anchor + shots)
+ r.Get("/albums", app.Wrap(albumHandler.List))
+ r.Post("/albums", app.Wrap(albumHandler.Create))
+ r.Get("/albums/{id}", app.Wrap(albumHandler.Get))
+ r.Delete("/albums/{id}", app.Wrap(albumHandler.Delete))
+ r.Post("/albums/{id}/anchor", app.Wrap(albumHandler.GenerateAnchor))
+ r.Post("/albums/{id}/shots", app.Wrap(albumHandler.GenerateAllShots))
+ r.Post("/albums/{id}/shots/{index}", app.Wrap(albumHandler.GenerateShot))
+ r.Delete("/albums/{id}/shots/{index}", app.Wrap(albumHandler.ResetShot))
+
+ // Persona generation (5-stage LLM + 20 images + 4 videos, all async)
+ r.Post("/persona/generate", app.Wrap(personaHandler.GeneratePersona))
+ })
+ })
+}
+
+// Dependencies holds all service dependencies for route registration.
+type Dependencies struct {
+ ExampleService *service.ExampleService
+ AuthService *service.AuthService
+ AlbumService *service.AlbumService
+ Queue queue.Producer
+ JobReader queue.JobReader
+ SSEHub *realtime.SSEHub
+ Store storage.Store
+ MediaRepo port.MediaRepository
+ EmailRenderer *emailpkg.Renderer
+}
diff --git a/services/persona-api/internal/api/spec.go b/services/persona-api/internal/api/spec.go
new file mode 100644
index 0000000..ae58a06
--- /dev/null
+++ b/services/persona-api/internal/api/spec.go
@@ -0,0 +1,112 @@
+package api
+
+import "git.threesix.ai/jordan/persona-community-2/pkg/openapi"
+
+// NewServiceSpec builds the OpenAPI specification for the persona-api service.
+func NewServiceSpec() *openapi.OpenAPISpec {
+ spec := openapi.NewOpenAPISpec("persona-api API", "1.0.0").
+ WithDescription("REST API for the persona-api service").
+ WithBearerSecurity("bearer", "JWT authentication token").
+ WithTag("Health", "Service health endpoints").
+ WithTag("Examples", "Example CRUD endpoints")
+
+ // Define reusable schemas
+ spec.WithSchema("Example", openapi.Object(map[string]openapi.Schema{
+ "id": openapi.UUID().WithDescription("Unique identifier"),
+ "name": openapi.String().WithDescription("Name of the example").WithExample("My Example"),
+ "description": openapi.String().WithDescription("Optional description").WithExample("A description"),
+ "created_at": openapi.DateTime().WithDescription("Creation timestamp"),
+ "updated_at": openapi.DateTime().WithDescription("Last update timestamp"),
+ }, "id", "name"))
+
+ spec.WithSchema("CreateExampleRequest", openapi.Object(map[string]openapi.Schema{
+ "name": openapi.StringWithMinMax(1, 100).WithDescription("Name of the example"),
+ "description": openapi.StringWithMinMax(0, 500).WithDescription("Optional description"),
+ }, "name"))
+
+ spec.WithSchema("UpdateExampleRequest", openapi.Object(map[string]openapi.Schema{
+ "name": openapi.StringWithMinMax(1, 100).WithDescription("Updated name"),
+ "description": openapi.StringWithMinMax(0, 500).WithDescription("Updated description"),
+ }))
+
+ // Health
+ spec.AddPath("/api/persona-api/health", "get", map[string]any{
+ "summary": "Health check",
+ "tags": []string{"Health"},
+ "responses": map[string]any{
+ "200": openapi.OpResponse("Service is healthy", openapi.Object(map[string]openapi.Schema{
+ "service": openapi.String(),
+ "status": openapi.String(),
+ })),
+ },
+ })
+
+ // List examples
+ spec.AddPath("/api/persona-api/examples", "get", map[string]any{
+ "summary": "List examples",
+ "description": "Returns a paginated list of examples.",
+ "tags": []string{"Examples"},
+ "parameters": []any{openapi.PageParam(), openapi.PerPageParam()},
+ "responses": map[string]any{
+ "200": openapi.OpResponse("Success", openapi.ResponseSchema(openapi.RefArray("Example"))),
+ },
+ })
+
+ // Get example
+ spec.AddPath("/api/persona-api/examples/{id}", "get", map[string]any{
+ "summary": "Get example by ID",
+ "tags": []string{"Examples"},
+ "parameters": []any{openapi.IDParam()},
+ "responses": map[string]any{
+ "200": openapi.OpResponse("Success", openapi.ResponseSchema(openapi.Ref("Example"))),
+ "404": openapi.OpResponse("Not found", openapi.ErrorResponseSchema()),
+ },
+ })
+
+ // Create example
+ spec.AddPath("/api/persona-api/examples", "post", map[string]any{
+ "summary": "Create example",
+ "description": "Creates a new example. Requires authentication.",
+ "tags": []string{"Examples"},
+ "security": []map[string][]string{{"bearer": {}}},
+ "requestBody": openapi.RequestBody(openapi.Ref("CreateExampleRequest"), true),
+ "responses": map[string]any{
+ "201": openapi.OpResponse("Created", openapi.ResponseSchema(openapi.Ref("Example"))),
+ "400": openapi.OpResponse("Bad request", openapi.ErrorResponseSchema()),
+ "401": openapi.OpResponse("Unauthorized", openapi.ErrorResponseSchema()),
+ "422": openapi.OpResponse("Validation error", openapi.ErrorResponseSchema()),
+ },
+ })
+
+ // Update example
+ spec.AddPath("/api/persona-api/examples/{id}", "put", map[string]any{
+ "summary": "Update example",
+ "description": "Updates an existing example. Requires authentication.",
+ "tags": []string{"Examples"},
+ "security": []map[string][]string{{"bearer": {}}},
+ "parameters": []any{openapi.IDParam()},
+ "requestBody": openapi.RequestBody(openapi.Ref("UpdateExampleRequest"), true),
+ "responses": map[string]any{
+ "200": openapi.OpResponse("Updated", openapi.ResponseSchema(openapi.Ref("Example"))),
+ "400": openapi.OpResponse("Bad request", openapi.ErrorResponseSchema()),
+ "401": openapi.OpResponse("Unauthorized", openapi.ErrorResponseSchema()),
+ "404": openapi.OpResponse("Not found", openapi.ErrorResponseSchema()),
+ },
+ })
+
+ // Delete example
+ spec.AddPath("/api/persona-api/examples/{id}", "delete", map[string]any{
+ "summary": "Delete example",
+ "description": "Deletes an example by ID. Requires authentication.",
+ "tags": []string{"Examples"},
+ "security": []map[string][]string{{"bearer": {}}},
+ "parameters": []any{openapi.IDParam()},
+ "responses": map[string]any{
+ "204": openapi.OpResponseNoContent(),
+ "401": openapi.OpResponse("Unauthorized", openapi.ErrorResponseSchema()),
+ "404": openapi.OpResponse("Not found", openapi.ErrorResponseSchema()),
+ },
+ })
+
+ return spec
+}
diff --git a/services/persona-api/internal/config/config.go b/services/persona-api/internal/config/config.go
new file mode 100644
index 0000000..7a19652
--- /dev/null
+++ b/services/persona-api/internal/config/config.go
@@ -0,0 +1,90 @@
+// Package config provides service-specific configuration.
+package config
+
+import (
+ "os"
+ "strings"
+
+ "git.threesix.ai/jordan/persona-community-2/pkg/config"
+)
+
+// Config extends the base config with persona-api-specific settings.
+type Config struct {
+ config.AppConfig
+ Server config.ServerConfig
+ Database config.DatabaseConfig
+ Logging config.LoggingConfig
+
+ // Auth
+ AuthEnabled bool
+ JWTSecret string
+ RegistrationEnabled bool
+
+ // Redis for cross-process SSE event delivery
+ RedisURL string
+
+ // Notify service for email delivery (OTP, magic links, password reset, etc.)
+ // When NotifyURL is empty, emails are logged to stdout (dev mode).
+ NotifyURL string
+ NotifyAPIKey string
+ NotifyHost string
+ NotifyFrom string
+
+ // Email branding — injected into every transactional email.
+ AppName string // APP_NAME, default: "persona-api"
+ AppURL string // APP_URL, default: ""
+ SupportEmail string // SUPPORT_EMAIL, default: NOTIFY_FROM value
+ LogoURL string // LOGO_URL, default: "" (hides logo area)
+ BrandColor string // BRAND_COLOR, default: "#6366f1"
+
+ // Dev mode seed user — seeded into the in-memory user store on startup so the
+ // developer's email is always available without re-registering after each restart.
+ // No effect when DATABASE_URL is set (production uses real persistence).
+ DevUserEmail string // DEV_USER_EMAIL, e.g. "you@example.com"
+ DevUserPassword string // DEV_USER_PASSWORD, default: "DevPassword1"
+}
+
+// Load reads configuration from environment variables.
+func Load() *Config {
+ regEnabled := true
+ if v := os.Getenv("REGISTRATION_ENABLED"); v != "" {
+ regEnabled = strings.EqualFold(v, "true")
+ }
+
+ notifyFrom := getEnvDefault("NOTIFY_FROM", "noreply@persona-community-2.com")
+
+ cfg := &Config{
+ AppConfig: config.ReadAppConfig(),
+ Server: config.ReadServerConfig(),
+ Database: config.ReadDatabaseConfig(),
+ Logging: config.ReadLoggingConfig(),
+
+ AuthEnabled: strings.EqualFold(os.Getenv("AUTH_ENABLED"), "true"),
+ JWTSecret: os.Getenv("JWT_SECRET"),
+ RegistrationEnabled: regEnabled,
+ RedisURL: os.Getenv("REDIS_URL"),
+
+ NotifyURL: os.Getenv("NOTIFY_URL"),
+ NotifyAPIKey: os.Getenv("NOTIFY_API_KEY"),
+ NotifyHost: os.Getenv("NOTIFY_HOST"),
+ NotifyFrom: notifyFrom,
+
+ AppName: getEnvDefault("APP_NAME", "persona-api"),
+ AppURL: os.Getenv("APP_URL"),
+ SupportEmail: getEnvDefault("SUPPORT_EMAIL", notifyFrom),
+ LogoURL: os.Getenv("LOGO_URL"),
+ BrandColor: getEnvDefault("BRAND_COLOR", "#6366f1"),
+
+ DevUserEmail: os.Getenv("DEV_USER_EMAIL"),
+ DevUserPassword: getEnvDefault("DEV_USER_PASSWORD", "DevPassword1"),
+ }
+
+ return cfg
+}
+
+func getEnvDefault(key, defaultVal string) string {
+ if v := os.Getenv(key); v != "" {
+ return v
+ }
+ return defaultVal
+}
diff --git a/services/persona-api/internal/domain/auth_code.go b/services/persona-api/internal/domain/auth_code.go
new file mode 100644
index 0000000..12cdf6f
--- /dev/null
+++ b/services/persona-api/internal/domain/auth_code.go
@@ -0,0 +1,32 @@
+package domain
+
+import "time"
+
+// AuthCodePurpose identifies what an auth code is used for.
+type AuthCodePurpose string
+
+const (
+ PurposeLoginOTP AuthCodePurpose = "login_otp"
+ PurposeMagicLink AuthCodePurpose = "magic_link"
+ PurposePasswordReset AuthCodePurpose = "password_reset"
+ PurposeEmailVerify AuthCodePurpose = "email_verify"
+)
+
+// AuthCode is a single-use, time-limited code for authentication flows.
+// Used by OTP login, magic links, password reset, and email verification.
+type AuthCode struct {
+ ID string
+ UserID *UserID // Nullable for magic link signup
+ Email string
+ Code string
+ Purpose AuthCodePurpose
+ ExpiresAt time.Time
+ UsedAt *time.Time
+ IPAddress string
+ CreatedAt time.Time
+}
+
+// IsValid returns true if the code has not been used and has not expired.
+func (c *AuthCode) IsValid() bool {
+ return c.UsedAt == nil && time.Now().Before(c.ExpiresAt)
+}
diff --git a/services/persona-api/internal/domain/errors.go b/services/persona-api/internal/domain/errors.go
new file mode 100644
index 0000000..a512fcc
--- /dev/null
+++ b/services/persona-api/internal/domain/errors.go
@@ -0,0 +1,36 @@
+// Package domain contains pure domain models with no external dependencies.
+// These types represent the core business concepts of the service.
+package domain
+
+import "errors"
+
+// Domain errors - these are business-level errors that should be translated
+// to appropriate HTTP status codes by the handler layer.
+var (
+ // ErrNotFound indicates a requested resource does not exist.
+ ErrNotFound = errors.New("not found")
+
+ // ErrExampleNotFound indicates the requested example does not exist.
+ ErrExampleNotFound = errors.New("example not found")
+
+ // ErrDuplicateExample indicates an example with the same name already exists.
+ ErrDuplicateExample = errors.New("example with this name already exists")
+
+ // ErrInvalidExampleName indicates the example name is invalid.
+ ErrInvalidExampleName = errors.New("invalid example name")
+
+ // Auth errors
+ ErrUserNotFound = errors.New("user not found")
+ ErrDuplicateEmail = errors.New("email already registered")
+ ErrInvalidCredentials = errors.New("invalid email or password")
+ ErrSessionNotFound = errors.New("session not found")
+ ErrSessionRevoked = errors.New("session has been revoked")
+ ErrInvalidAuthCode = errors.New("invalid or expired code")
+ ErrExpiredAuthCode = errors.New("code has expired")
+ ErrWeakPassword = errors.New("password does not meet requirements")
+ ErrUserSuspended = errors.New("account is suspended")
+ ErrRegistrationDisabled = errors.New("registration is disabled")
+ ErrNameTooLong = errors.New("name exceeds maximum length")
+ ErrEmailTooLong = errors.New("email exceeds maximum length")
+ ErrInvalidAvatarURL = errors.New("avatar URL must use http or https")
+)
diff --git a/services/persona-api/internal/domain/example.go b/services/persona-api/internal/domain/example.go
new file mode 100644
index 0000000..4ee48e9
--- /dev/null
+++ b/services/persona-api/internal/domain/example.go
@@ -0,0 +1,89 @@
+package domain
+
+import (
+ "time"
+ "unicode/utf8"
+)
+
+// ExampleID is a strongly-typed identifier for examples.
+type ExampleID string
+
+// String returns the string representation of the ID.
+func (id ExampleID) String() string {
+ return string(id)
+}
+
+// IsZero returns true if the ID is empty.
+func (id ExampleID) IsZero() bool {
+ return id == ""
+}
+
+// Example name constraints.
+const (
+ MinExampleNameLen = 1
+ MaxExampleNameLen = 100
+ MaxDescriptionLen = 500
+)
+
+// Example represents an example domain entity.
+// This is a pure domain model with no external dependencies.
+type Example struct {
+ ID ExampleID
+ Name string
+ Description string
+ CreatedAt time.Time
+ UpdatedAt time.Time
+}
+
+// NewExample creates a new Example with validation.
+// Returns ErrInvalidExampleName if the name is invalid.
+func NewExample(id ExampleID, name, description string) (*Example, error) {
+ if err := validateExampleName(name); err != nil {
+ return nil, err
+ }
+ if err := validateDescription(description); err != nil {
+ return nil, err
+ }
+
+ now := time.Now().UTC()
+ return &Example{
+ ID: id,
+ Name: name,
+ Description: description,
+ CreatedAt: now,
+ UpdatedAt: now,
+ }, nil
+}
+
+// Update modifies the example's mutable fields with validation.
+// Returns ErrInvalidExampleName if the name is invalid.
+func (e *Example) Update(name, description string) error {
+ if err := validateExampleName(name); err != nil {
+ return err
+ }
+ if err := validateDescription(description); err != nil {
+ return err
+ }
+
+ e.Name = name
+ e.Description = description
+ e.UpdatedAt = time.Now().UTC()
+ return nil
+}
+
+// validateExampleName validates an example name.
+func validateExampleName(name string) error {
+ length := utf8.RuneCountInString(name)
+ if length < MinExampleNameLen || length > MaxExampleNameLen {
+ return ErrInvalidExampleName
+ }
+ return nil
+}
+
+// validateDescription validates a description.
+func validateDescription(desc string) error {
+ if utf8.RuneCountInString(desc) > MaxDescriptionLen {
+ return ErrInvalidExampleName
+ }
+ return nil
+}
diff --git a/services/persona-api/internal/domain/media.go b/services/persona-api/internal/domain/media.go
new file mode 100644
index 0000000..22a4ee4
--- /dev/null
+++ b/services/persona-api/internal/domain/media.go
@@ -0,0 +1,27 @@
+package domain
+
+import "time"
+
+// MediaObjectID is a typed media object identifier with prefix "med_".
+type MediaObjectID string
+
+// MediaObject tracks a stored media file with ownership and metadata.
+// The actual file is stored in GCS (production) or MemoryStore (dev).
+// This record enables querying, soft deletes, and provenance tracking.
+type MediaObject struct {
+ ID MediaObjectID
+ UserID UserID
+ Path string // Storage path (e.g., "media/usr_123/uuid/photo.png")
+ Filename string // Original filename
+ ContentType string
+ Size int64
+ GenerationJobID string // Non-empty if created by AI generation
+ DeletedAt *time.Time
+ CreatedAt time.Time
+ UpdatedAt time.Time
+}
+
+// IsDeleted returns true if the media object has been soft-deleted.
+func (m *MediaObject) IsDeleted() bool {
+ return m.DeletedAt != nil
+}
diff --git a/services/persona-api/internal/domain/session.go b/services/persona-api/internal/domain/session.go
new file mode 100644
index 0000000..81e89ae
--- /dev/null
+++ b/services/persona-api/internal/domain/session.go
@@ -0,0 +1,25 @@
+package domain
+
+import "time"
+
+// SessionID is a typed session identifier with prefix "ses_".
+type SessionID string
+
+// Session tracks a user login with device and location information.
+// The session ID is embedded in the JWT token for revocation support.
+type Session struct {
+ ID SessionID
+ UserID UserID
+ IPAddress string
+ UserAgent string
+ DeviceLabel string
+ LastActiveAt time.Time
+ ExpiresAt time.Time
+ RevokedAt *time.Time
+ CreatedAt time.Time
+}
+
+// IsActive returns true if the session has not been revoked and has not expired.
+func (s *Session) IsActive() bool {
+ return s.RevokedAt == nil && time.Now().Before(s.ExpiresAt)
+}
diff --git a/services/persona-api/internal/domain/user.go b/services/persona-api/internal/domain/user.go
new file mode 100644
index 0000000..2b24d21
--- /dev/null
+++ b/services/persona-api/internal/domain/user.go
@@ -0,0 +1,52 @@
+package domain
+
+import "time"
+
+// UserID is a typed user identifier with prefix "usr_".
+type UserID string
+
+// UserStatus represents the account state.
+type UserStatus string
+
+const (
+ UserStatusActive UserStatus = "active"
+ UserStatusSuspended UserStatus = "suspended"
+ UserStatusDeactivated UserStatus = "deactivated"
+)
+
+// User is the full domain model for a registered user.
+// This is the database-backed identity, separate from auth.User which is the
+// lightweight JWT-derived identity carried in request context.
+type User struct {
+ ID UserID
+ Email string
+ EmailVerified bool
+ Name string
+ AvatarURL string
+ Status UserStatus
+ Roles []string
+ LastLoginAt *time.Time
+ CreatedAt time.Time
+ UpdatedAt time.Time
+}
+
+// Validation constants for user fields.
+const (
+ MaxNameLen = 100
+ MaxEmailLen = 254 // RFC 5321
+)
+
+// NewUser creates a new user with default values.
+func NewUser(id UserID, email, name string) *User {
+ now := time.Now()
+ return &User{
+ ID: id,
+ Email: email,
+ EmailVerified: false,
+ Name: name,
+ Status: UserStatusActive,
+ Roles: []string{"user"},
+ CreatedAt: now,
+ UpdatedAt: now,
+ }
+}
diff --git a/services/persona-api/internal/email/embed.go b/services/persona-api/internal/email/embed.go
new file mode 100644
index 0000000..a873619
--- /dev/null
+++ b/services/persona-api/internal/email/embed.go
@@ -0,0 +1,8 @@
+// Package email provides embedded email templates for persona-api transactional emails.
+// The TemplateFS is passed to pkg/email.NewRendererFromFS at startup.
+package email
+
+import "embed"
+
+//go:embed all:templates
+var TemplateFS embed.FS
diff --git a/services/persona-api/internal/email/renderer_test.go b/services/persona-api/internal/email/renderer_test.go
new file mode 100644
index 0000000..a732ad2
--- /dev/null
+++ b/services/persona-api/internal/email/renderer_test.go
@@ -0,0 +1,202 @@
+package email_test
+
+import (
+ "strings"
+ "testing"
+
+ emailpkg "git.threesix.ai/jordan/persona-community-2/pkg/email"
+ "git.threesix.ai/jordan/persona-community-2/services/persona-api/internal/email"
+)
+
+var testBrand = emailpkg.BrandConfig{
+ AppName: "Test App",
+ AppURL: "https://example.com",
+ SupportEmail: "support@example.com",
+ PrimaryColor: "#6366f1",
+}
+
+func newTestRenderer(t *testing.T) *emailpkg.Renderer {
+ t.Helper()
+ r, err := emailpkg.NewRendererFromFS(email.TemplateFS, "templates", testBrand)
+ if err != nil {
+ t.Fatalf("NewRendererFromFS: %v", err)
+ }
+ return r
+}
+
+func TestRendererLoads(t *testing.T) {
+ r := newTestRenderer(t)
+
+ purposes := r.Purposes()
+ want := []string{"email_verify", "login_otp", "magic_link", "password_reset", "welcome"}
+ if len(purposes) != len(want) {
+ t.Fatalf("expected %d purposes, got %d: %v", len(want), len(purposes), purposes)
+ }
+ for i, p := range want {
+ if purposes[i] != p {
+ t.Errorf("purpose[%d]: want %q, got %q", i, p, purposes[i])
+ }
+ }
+}
+
+func TestRenderLoginOTP(t *testing.T) {
+ r := newTestRenderer(t)
+ out, err := r.Render("login_otp", emailpkg.EmailContext{
+ Code: "482916",
+ ExpiresIn: 10,
+ Purpose: "sign in",
+ })
+ if err != nil {
+ t.Fatalf("Render login_otp: %v", err)
+ }
+ if out.Subject == "" {
+ t.Error("Subject is empty")
+ }
+ if !strings.Contains(out.Subject, "Test App") {
+ t.Errorf("Subject %q does not contain app name", out.Subject)
+ }
+ if !strings.Contains(out.HTML, "482916") {
+ t.Error("HTML does not contain OTP code")
+ }
+ if !strings.Contains(out.HTML, "code-box") {
+ t.Error("HTML does not contain code-box element")
+ }
+ if out.PlainText == "" {
+ t.Error("PlainText is empty")
+ }
+ if !strings.Contains(out.PlainText, "482916") {
+ t.Error("PlainText does not contain OTP code")
+ }
+ if out.Preheader == "" {
+ t.Error("Preheader is empty")
+ }
+}
+
+func TestRenderMagicLink(t *testing.T) {
+ r := newTestRenderer(t)
+ out, err := r.Render("magic_link", emailpkg.EmailContext{
+ ActionURL: "https://example.com/auth/verify?token=abc123",
+ ButtonText: "Sign In \u2192",
+ ExpiresIn: 15,
+ })
+ if err != nil {
+ t.Fatalf("Render magic_link: %v", err)
+ }
+ if !strings.Contains(out.HTML, "Sign In") {
+ t.Error("HTML does not contain button text")
+ }
+ if !strings.Contains(out.HTML, "auth/verify") {
+ t.Error("HTML does not contain action URL")
+ }
+ if out.PlainText == "" {
+ t.Error("PlainText is empty")
+ }
+}
+
+func TestRenderPasswordReset(t *testing.T) {
+ r := newTestRenderer(t)
+ out, err := r.Render("password_reset", emailpkg.EmailContext{
+ ActionURL: "https://example.com/auth/reset?token=xyz789",
+ ButtonText: "Reset Password \u2192",
+ ExpiresIn: 60,
+ })
+ if err != nil {
+ t.Fatalf("Render password_reset: %v", err)
+ }
+ if !strings.Contains(out.HTML, "Reset Password") {
+ t.Error("HTML does not contain button text")
+ }
+ if !strings.Contains(out.Subject, "Reset") {
+ t.Errorf("Subject %q does not mention reset", out.Subject)
+ }
+}
+
+func TestRenderVerifyEmail(t *testing.T) {
+ r := newTestRenderer(t)
+ out, err := r.Render("email_verify", emailpkg.EmailContext{
+ Code: "738201",
+ ExpiresIn: 30,
+ Purpose: "verify your email",
+ })
+ if err != nil {
+ t.Fatalf("Render email_verify: %v", err)
+ }
+ if !strings.Contains(out.HTML, "738201") {
+ t.Error("HTML does not contain verification code")
+ }
+ if !strings.Contains(out.HTML, "code-box") {
+ t.Error("HTML does not contain code-box element")
+ }
+}
+
+func TestRenderWelcome(t *testing.T) {
+ r := newTestRenderer(t)
+ out, err := r.Render("welcome", emailpkg.EmailContext{
+ ActionURL: "https://example.com/dashboard",
+ ButtonText: "Get Started \u2192",
+ Name: "Jordan",
+ })
+ if err != nil {
+ t.Fatalf("Render welcome: %v", err)
+ }
+ if !strings.Contains(out.HTML, "Jordan") {
+ t.Error("HTML does not contain user name")
+ }
+ if !strings.Contains(out.HTML, "Welcome") {
+ t.Error("HTML does not contain welcome heading")
+ }
+}
+
+func TestBrandColorInjection(t *testing.T) {
+ r := newTestRenderer(t)
+ out, err := r.Render("login_otp", emailpkg.EmailContext{
+ Code: "123456",
+ ExpiresIn: 10,
+ })
+ if err != nil {
+ t.Fatalf("Render: %v", err)
+ }
+ // Brand primary color should appear in the inlined styles.
+ if !strings.Contains(out.HTML, "#6366f1") {
+ t.Error("HTML does not contain brand color #6366f1")
+ }
+}
+
+func TestUnknownPurposeReturnsError(t *testing.T) {
+ r := newTestRenderer(t)
+ _, err := r.Render("nonexistent_type", emailpkg.EmailContext{})
+ if err == nil {
+ t.Error("expected error for unknown purpose, got nil")
+ }
+}
+
+func TestAllTemplatesHaveSubjectAndPreheader(t *testing.T) {
+ r := newTestRenderer(t)
+ contexts := map[string]emailpkg.EmailContext{
+ "login_otp": {Code: "111111", ExpiresIn: 10},
+ "magic_link": {ActionURL: "https://example.com/auth", ButtonText: "Sign In", ExpiresIn: 15},
+ "password_reset": {ActionURL: "https://example.com/reset", ButtonText: "Reset", ExpiresIn: 60},
+ "email_verify": {Code: "222222", ExpiresIn: 30},
+ "welcome": {ActionURL: "https://example.com", ButtonText: "Get Started", Name: "Alex"},
+ }
+ for _, purpose := range r.Purposes() {
+ ctx := contexts[purpose]
+ out, err := r.Render(purpose, ctx)
+ if err != nil {
+ t.Errorf("%s: render error: %v", purpose, err)
+ continue
+ }
+ if out.Subject == "" {
+ t.Errorf("%s: Subject is empty", purpose)
+ }
+ if out.Preheader == "" {
+ t.Errorf("%s: Preheader is empty", purpose)
+ }
+ if out.HTML == "" {
+ t.Errorf("%s: HTML is empty", purpose)
+ }
+ if out.PlainText == "" {
+ t.Errorf("%s: PlainText is empty", purpose)
+ }
+ }
+}
diff --git a/services/persona-api/internal/email/templates/_button.html b/services/persona-api/internal/email/templates/_button.html
new file mode 100644
index 0000000..bbfd435
--- /dev/null
+++ b/services/persona-api/internal/email/templates/_button.html
@@ -0,0 +1,4 @@
+{{define "button"}}
+{{end}}
diff --git a/services/persona-api/internal/email/templates/_code_box.html b/services/persona-api/internal/email/templates/_code_box.html
new file mode 100644
index 0000000..f0ecadc
--- /dev/null
+++ b/services/persona-api/internal/email/templates/_code_box.html
@@ -0,0 +1,4 @@
+{{define "code_box"}}
+ {{.Code}}
+
+{{end}}
diff --git a/services/persona-api/internal/email/templates/_footer.html b/services/persona-api/internal/email/templates/_footer.html
new file mode 100644
index 0000000..0c98740
--- /dev/null
+++ b/services/persona-api/internal/email/templates/_footer.html
@@ -0,0 +1,7 @@
+{{define "footer"}}
+{{end}}
diff --git a/services/persona-api/internal/email/templates/_header.html b/services/persona-api/internal/email/templates/_header.html
new file mode 100644
index 0000000..4e59b9e
--- /dev/null
+++ b/services/persona-api/internal/email/templates/_header.html
@@ -0,0 +1,8 @@
+{{define "header"}}
+{{end}}
diff --git a/services/persona-api/internal/email/templates/_layout.html b/services/persona-api/internal/email/templates/_layout.html
new file mode 100644
index 0000000..1db0a29
--- /dev/null
+++ b/services/persona-api/internal/email/templates/_layout.html
@@ -0,0 +1,83 @@
+{{define "layout"}}
+
+
+
+
+
+
+
+ {{.AppName}}
+
+
+
+
+
+
+
+ {{template "header" .}}
+
+ {{template "body" .}}
+
+ {{template "footer" .}}
+
+
Hi there,
+Enter this code to verify your email address for {{.AppName}}:
+
+{{template "code_box" .}}
+
+This code expires in {{.ExpiresIn}} minutes. If you didn't request this, you can safely ignore this email.
+{{end}}
diff --git a/services/persona-api/internal/email/templates/email_verify/meta.yaml b/services/persona-api/internal/email/templates/email_verify/meta.yaml
new file mode 100644
index 0000000..8978ac6
--- /dev/null
+++ b/services/persona-api/internal/email/templates/email_verify/meta.yaml
@@ -0,0 +1,4 @@
+purpose: email_verify
+category: transactional
+subject: "Verify your {{.AppName}} email address"
+preheader: "{{.Code}} is your email verification code — expires in {{.ExpiresIn}} minutes."
diff --git a/services/persona-api/internal/email/templates/login_otp/body.html b/services/persona-api/internal/email/templates/login_otp/body.html
new file mode 100644
index 0000000..5d4315d
--- /dev/null
+++ b/services/persona-api/internal/email/templates/login_otp/body.html
@@ -0,0 +1,8 @@
+{{define "body"}}
+Hi there,
+Here is your sign in code for {{.AppName}}:
+
+{{template "code_box" .}}
+
+This code expires in {{.ExpiresIn}} minutes. If you didn't request this, you can safely ignore this email — your account is secure.
+{{end}}
diff --git a/services/persona-api/internal/email/templates/login_otp/meta.yaml b/services/persona-api/internal/email/templates/login_otp/meta.yaml
new file mode 100644
index 0000000..1eac2d6
--- /dev/null
+++ b/services/persona-api/internal/email/templates/login_otp/meta.yaml
@@ -0,0 +1,4 @@
+purpose: login_otp
+category: transactional
+subject: "Your {{.AppName}} sign in code"
+preheader: "{{.Code}} is your sign in code — expires in {{.ExpiresIn}} minutes."
diff --git a/services/persona-api/internal/email/templates/magic_link/body.html b/services/persona-api/internal/email/templates/magic_link/body.html
new file mode 100644
index 0000000..9e7044a
--- /dev/null
+++ b/services/persona-api/internal/email/templates/magic_link/body.html
@@ -0,0 +1,13 @@
+{{define "body"}}
+Sign in to {{.AppName}}
+Click the button below to sign in. This link expires in {{.ExpiresIn}} minutes.
+
+{{template "button" .}}
+
+Or copy this link into your browser:
+{{.ActionURL}}
+
+
+
+If you didn't request this sign-in link, you can safely ignore this email — your account is secure.
+{{end}}
diff --git a/services/persona-api/internal/email/templates/magic_link/meta.yaml b/services/persona-api/internal/email/templates/magic_link/meta.yaml
new file mode 100644
index 0000000..08ebb3d
--- /dev/null
+++ b/services/persona-api/internal/email/templates/magic_link/meta.yaml
@@ -0,0 +1,4 @@
+purpose: magic_link
+category: transactional
+subject: "Sign in to {{.AppName}}"
+preheader: "Click to sign in to your {{.AppName}} account — link expires in {{.ExpiresIn}} minutes."
diff --git a/services/persona-api/internal/email/templates/password_reset/body.html b/services/persona-api/internal/email/templates/password_reset/body.html
new file mode 100644
index 0000000..e2bdcb6
--- /dev/null
+++ b/services/persona-api/internal/email/templates/password_reset/body.html
@@ -0,0 +1,13 @@
+{{define "body"}}
+Reset your password
+Click the button below to choose a new password. This link expires in {{.ExpiresIn}} minutes.
+
+{{template "button" .}}
+
+Or copy this link into your browser:
+{{.ActionURL}}
+
+
+
+If you didn't request a password reset, you can safely ignore this email — your password won't change.
+{{end}}
diff --git a/services/persona-api/internal/email/templates/password_reset/meta.yaml b/services/persona-api/internal/email/templates/password_reset/meta.yaml
new file mode 100644
index 0000000..67e8b7f
--- /dev/null
+++ b/services/persona-api/internal/email/templates/password_reset/meta.yaml
@@ -0,0 +1,4 @@
+purpose: password_reset
+category: transactional
+subject: "Reset your {{.AppName}} password"
+preheader: "You requested a password reset for {{.AppName}} — link expires in {{.ExpiresIn}} minutes."
diff --git a/services/persona-api/internal/email/templates/welcome/body.html b/services/persona-api/internal/email/templates/welcome/body.html
new file mode 100644
index 0000000..2eac89b
--- /dev/null
+++ b/services/persona-api/internal/email/templates/welcome/body.html
@@ -0,0 +1,10 @@
+{{define "body"}}
+Welcome to {{.AppName}}{{if .Name}}, {{.Name}}{{end}}!
+Your account is ready. Start exploring everything {{.AppName}} has to offer.
+
+{{template "button" .}}
+
+{{- if .SupportEmail}}
+Have questions? Reach us at {{.SupportEmail}}.
+{{- end}}
+{{end}}
diff --git a/services/persona-api/internal/email/templates/welcome/meta.yaml b/services/persona-api/internal/email/templates/welcome/meta.yaml
new file mode 100644
index 0000000..4f9bc92
--- /dev/null
+++ b/services/persona-api/internal/email/templates/welcome/meta.yaml
@@ -0,0 +1,4 @@
+purpose: welcome
+category: transactional
+subject: "Welcome to {{.AppName}}"
+preheader: "Your {{.AppName}} account is ready. Get started today."
diff --git a/services/persona-api/internal/port/album.go b/services/persona-api/internal/port/album.go
new file mode 100644
index 0000000..16e8c9a
--- /dev/null
+++ b/services/persona-api/internal/port/album.go
@@ -0,0 +1,34 @@
+package port
+
+import (
+ "context"
+
+ "git.threesix.ai/jordan/persona-community-2/pkg/album"
+)
+
+// AlbumRepository defines persistence operations for albums.
+// It extends album.AlbumUpdater so implementations satisfy both interfaces.
+type AlbumRepository interface {
+ album.AlbumUpdater
+
+ // Create persists a new album. Sets ID, CreatedAt, UpdatedAt.
+ Create(ctx context.Context, a *album.Album) error
+
+ // Get returns an album by ID. Returns ErrAlbumNotFound if not found.
+ Get(ctx context.Context, id album.AlbumID, userID string) (*album.Album, error)
+
+ // List returns all albums for a user, ordered by CreatedAt DESC.
+ List(ctx context.Context, userID string) ([]album.Album, error)
+
+ // Delete removes an album and all its shots. Does NOT delete stored images.
+ Delete(ctx context.Context, id album.AlbumID, userID string) error
+
+ // ResetShot clears a shot's ImageURL, JobID, Error, and sets Status to pending.
+ ResetShot(ctx context.Context, id album.AlbumID, userID string, shotIndex int) error
+
+ // UpdateAnchorJobID sets the AnchorJobID when the anchor generation job is enqueued.
+ UpdateAnchorJobID(ctx context.Context, id album.AlbumID, userID, jobID string) error
+
+ // UpdateShotJobID sets the shot's JobID when a shot generation job is enqueued.
+ UpdateShotJobID(ctx context.Context, id album.AlbumID, userID string, shotIndex int, jobID string) error
+}
diff --git a/services/persona-api/internal/port/auth_code.go b/services/persona-api/internal/port/auth_code.go
new file mode 100644
index 0000000..f5e7f56
--- /dev/null
+++ b/services/persona-api/internal/port/auth_code.go
@@ -0,0 +1,24 @@
+package port
+
+import (
+ "context"
+
+ "git.threesix.ai/jordan/persona-community-2/services/persona-api/internal/domain"
+)
+
+// AuthCodeRepository defines the interface for auth code persistence.
+type AuthCodeRepository interface {
+ // Create persists a new auth code.
+ Create(ctx context.Context, code *domain.AuthCode) error
+
+ // FindValid returns an unused, non-expired code matching the criteria.
+ // Returns domain.ErrInvalidAuthCode if no valid code exists.
+ FindValid(ctx context.Context, email string, code string, purpose domain.AuthCodePurpose) (*domain.AuthCode, error)
+
+ // MarkUsed sets the used_at timestamp on a code, making it single-use.
+ MarkUsed(ctx context.Context, id string) error
+
+ // DeleteExpired removes codes that have passed their expiry time.
+ // Returns the number of codes deleted.
+ DeleteExpired(ctx context.Context) (int, error)
+}
diff --git a/services/persona-api/internal/port/email.go b/services/persona-api/internal/port/email.go
new file mode 100644
index 0000000..9b54741
--- /dev/null
+++ b/services/persona-api/internal/port/email.go
@@ -0,0 +1,11 @@
+package port
+
+import "context"
+
+// EmailSender sends emails for authentication flows (OTP, magic link, password reset, etc.).
+type EmailSender interface {
+ // SendAuthCode sends an authentication code to the given email.
+ // purpose identifies the flow (e.g. "login_otp", "magic_link", "password_reset", "email_verify").
+ // code is the token or OTP to include in the email.
+ SendAuthCode(ctx context.Context, email, code, purpose string) error
+}
diff --git a/services/persona-api/internal/port/example.go b/services/persona-api/internal/port/example.go
new file mode 100644
index 0000000..120ede6
--- /dev/null
+++ b/services/persona-api/internal/port/example.go
@@ -0,0 +1,37 @@
+// Package port defines interfaces (ports) for external dependencies.
+// These interfaces define the contracts between the application core and
+// infrastructure adapters, enabling testability and flexibility.
+package port
+
+import (
+ "context"
+
+ "git.threesix.ai/jordan/persona-community-2/services/persona-api/internal/domain"
+)
+
+// ExampleRepository defines the interface for example persistence operations.
+// Implementations may use databases, in-memory storage, or external services.
+type ExampleRepository interface {
+ // List returns all examples.
+ List(ctx context.Context) ([]domain.Example, error)
+
+ // Get returns an example by ID.
+ // Returns domain.ErrExampleNotFound if not found.
+ Get(ctx context.Context, id domain.ExampleID) (*domain.Example, error)
+
+ // Create stores a new example.
+ // The example must have a valid ID set.
+ Create(ctx context.Context, example *domain.Example) error
+
+ // Update modifies an existing example.
+ // Returns domain.ErrExampleNotFound if not found.
+ Update(ctx context.Context, example *domain.Example) error
+
+ // Delete removes an example by ID.
+ // Returns domain.ErrExampleNotFound if not found.
+ Delete(ctx context.Context, id domain.ExampleID) error
+
+ // ExistsByName checks if an example with the given name exists.
+ // Used for duplicate detection.
+ ExistsByName(ctx context.Context, name string) (bool, error)
+}
diff --git a/services/persona-api/internal/port/media.go b/services/persona-api/internal/port/media.go
new file mode 100644
index 0000000..a8e578d
--- /dev/null
+++ b/services/persona-api/internal/port/media.go
@@ -0,0 +1,38 @@
+package port
+
+import (
+ "context"
+
+ "git.threesix.ai/jordan/persona-community-2/services/persona-api/internal/domain"
+)
+
+// MediaRepository defines the interface for media metadata persistence.
+type MediaRepository interface {
+ // Create persists a new media object record.
+ Create(ctx context.Context, obj *domain.MediaObject) error
+
+ // Get returns a media object by ID. Returns domain.ErrNotFound if not found or soft-deleted.
+ Get(ctx context.Context, id domain.MediaObjectID) (*domain.MediaObject, error)
+
+ // ListByUser returns non-deleted media objects for a user, ordered by created_at DESC.
+ ListByUser(ctx context.Context, userID domain.UserID, opts ListMediaOptions) ([]domain.MediaObject, int, error)
+
+ // SoftDelete marks a media object as deleted without removing it.
+ SoftDelete(ctx context.Context, id domain.MediaObjectID) error
+
+ // HardDelete permanently removes a media object record.
+ HardDelete(ctx context.Context, id domain.MediaObjectID) error
+
+ // GetByPath returns a media object by its storage path. Returns domain.ErrNotFound if not found.
+ GetByPath(ctx context.Context, path string) (*domain.MediaObject, error)
+}
+
+// ListMediaOptions controls filtering and pagination for media queries.
+type ListMediaOptions struct {
+ // ContentTypePrefix filters by MIME type prefix (e.g., "image/", "video/").
+ ContentTypePrefix string
+ // Limit is the maximum number of results (0 = default 50).
+ Limit int
+ // Offset is the pagination offset.
+ Offset int
+}
diff --git a/services/persona-api/internal/port/session.go b/services/persona-api/internal/port/session.go
new file mode 100644
index 0000000..e9dd778
--- /dev/null
+++ b/services/persona-api/internal/port/session.go
@@ -0,0 +1,33 @@
+package port
+
+import (
+ "context"
+
+ "git.threesix.ai/jordan/persona-community-2/services/persona-api/internal/domain"
+)
+
+// SessionRepository defines the interface for session persistence.
+type SessionRepository interface {
+ // Create persists a new session record.
+ Create(ctx context.Context, session *domain.Session) error
+
+ // Get returns a session by ID. Returns domain.ErrSessionNotFound if not found.
+ Get(ctx context.Context, id domain.SessionID) (*domain.Session, error)
+
+ // ListByUser returns all active (non-revoked) sessions for a user.
+ ListByUser(ctx context.Context, userID domain.UserID) ([]domain.Session, error)
+
+ // UpdateLastActive updates the last_active_at timestamp for a session.
+ UpdateLastActive(ctx context.Context, id domain.SessionID) error
+
+ // Revoke marks a session as revoked by setting revoked_at.
+ Revoke(ctx context.Context, id domain.SessionID) error
+
+ // RevokeAllForUser revokes all sessions for a user.
+ // If exceptID is non-nil, that session is kept active.
+ RevokeAllForUser(ctx context.Context, userID domain.UserID, exceptID *domain.SessionID) error
+
+ // DeleteExpired removes sessions that have passed their expiry time.
+ // Returns the number of sessions deleted.
+ DeleteExpired(ctx context.Context) (int, error)
+}
diff --git a/services/persona-api/internal/port/user.go b/services/persona-api/internal/port/user.go
new file mode 100644
index 0000000..b91578b
--- /dev/null
+++ b/services/persona-api/internal/port/user.go
@@ -0,0 +1,51 @@
+package port
+
+import (
+ "context"
+
+ "git.threesix.ai/jordan/persona-community-2/services/persona-api/internal/domain"
+)
+
+// UserRepository defines the interface for user persistence.
+type UserRepository interface {
+ // Create persists a new user.
+ Create(ctx context.Context, user *domain.User) error
+
+ // Get returns a user by ID. Returns domain.ErrUserNotFound if not found.
+ Get(ctx context.Context, id domain.UserID) (*domain.User, error)
+
+ // GetByEmail returns a user by email. Returns domain.ErrUserNotFound if not found.
+ GetByEmail(ctx context.Context, email string) (*domain.User, error)
+
+ // Update persists changes to an existing user.
+ Update(ctx context.Context, user *domain.User) error
+
+ // UpdateLastLogin sets the last_login_at timestamp.
+ UpdateLastLogin(ctx context.Context, id domain.UserID) error
+
+ // ExistsByEmail returns true if a user with the given email exists.
+ ExistsByEmail(ctx context.Context, email string) (bool, error)
+
+ // Password operations (separate from user CRUD because OAuth-only users have no password)
+
+ // SetPassword stores a bcrypt hash for a user. Creates or replaces existing.
+ SetPassword(ctx context.Context, userID domain.UserID, hash string) error
+
+ // GetPasswordHash returns the bcrypt hash for a user.
+ // Returns empty string and nil error if user has no password set.
+ GetPasswordHash(ctx context.Context, userID domain.UserID) (string, error)
+
+ // HasPassword returns true if the user has a password set.
+ HasPassword(ctx context.Context, userID domain.UserID) (bool, error)
+
+ // Role operations
+
+ // AddRole grants a role to a user. No-op if already granted.
+ AddRole(ctx context.Context, userID domain.UserID, role string) error
+
+ // RemoveRole revokes a role from a user. No-op if not granted.
+ RemoveRole(ctx context.Context, userID domain.UserID, role string) error
+
+ // GetRoles returns all roles for a user.
+ GetRoles(ctx context.Context, userID domain.UserID) ([]string, error)
+}
diff --git a/services/persona-api/internal/service/album.go b/services/persona-api/internal/service/album.go
new file mode 100644
index 0000000..b7d136c
--- /dev/null
+++ b/services/persona-api/internal/service/album.go
@@ -0,0 +1,186 @@
+package service
+
+import (
+ "context"
+ "fmt"
+
+ "git.threesix.ai/jordan/persona-community-2/pkg/album"
+ "git.threesix.ai/jordan/persona-community-2/pkg/logging"
+ "git.threesix.ai/jordan/persona-community-2/pkg/queue"
+ "git.threesix.ai/jordan/persona-community-2/services/persona-api/internal/port"
+)
+
+// AlbumService handles album creation, retrieval, and generation orchestration.
+// All generation is async: service enqueues jobs and returns immediately.
+// Results arrive via SSE on the user: channel.
+type AlbumService struct {
+ albums port.AlbumRepository
+ queue queue.Producer
+ logger *logging.Logger
+}
+
+// NewAlbumService creates a new AlbumService.
+func NewAlbumService(albums port.AlbumRepository, q queue.Producer, logger *logging.Logger) *AlbumService {
+ return &AlbumService{
+ albums: albums,
+ queue: q,
+ logger: logger.WithComponent("AlbumService"),
+ }
+}
+
+// Create creates a new album with the given shots and persists it.
+// Shots are provided as ShotTemplate slices (Label + Direction).
+func (s *AlbumService) Create(ctx context.Context, userID, name, subjectDesc string, shots []album.ShotTemplate) (*album.Album, error) {
+ if name == "" {
+ return nil, fmt.Errorf("album name is required")
+ }
+ if subjectDesc == "" {
+ return nil, fmt.Errorf("subject description is required")
+ }
+ if len(shots) == 0 {
+ return nil, fmt.Errorf("at least one shot is required")
+ }
+ if len(shots) > 20 {
+ return nil, fmt.Errorf("maximum 20 shots per album")
+ }
+
+ shotList := make([]album.Shot, len(shots))
+ for i, tmpl := range shots {
+ shotList[i] = album.Shot{
+ Index: i,
+ Label: tmpl.Label,
+ Direction: tmpl.Direction,
+ Status: album.ShotPending,
+ }
+ }
+
+ a := &album.Album{
+ ID: album.AlbumID("alb_" + generateID()),
+ UserID: userID,
+ Name: name,
+ SubjectDesc: subjectDesc,
+ Shots: shotList,
+ }
+
+ if err := s.albums.Create(ctx, a); err != nil {
+ return nil, fmt.Errorf("create album: %w", err)
+ }
+
+ s.logger.Info("album created", "album_id", string(a.ID), "user_id", userID, "shots", len(shotList))
+ return a, nil
+}
+
+// Get returns an album by ID, enforcing user ownership.
+func (s *AlbumService) Get(ctx context.Context, id album.AlbumID, userID string) (*album.Album, error) {
+ a, err := s.albums.Get(ctx, id, userID)
+ if err != nil {
+ return nil, fmt.Errorf("album not found: %w", err)
+ }
+ return a, nil
+}
+
+// List returns all albums for a user.
+func (s *AlbumService) List(ctx context.Context, userID string) ([]album.Album, error) {
+ return s.albums.List(ctx, userID)
+}
+
+// Delete removes an album. Does NOT delete stored images.
+func (s *AlbumService) Delete(ctx context.Context, id album.AlbumID, userID string) error {
+ return s.albums.Delete(ctx, id, userID)
+}
+
+// GenerateAnchor enqueues an anchor generation job for an album.
+// Returns the job ID. Result arrives via album_anchor_complete SSE event.
+func (s *AlbumService) GenerateAnchor(ctx context.Context, id album.AlbumID, userID string) (string, error) {
+ a, err := s.albums.Get(ctx, id, userID)
+ if err != nil {
+ return "", fmt.Errorf("album not found: %w", err)
+ }
+
+ jobID, err := s.queue.Enqueue(ctx, "generate_anchor", map[string]any{
+ "albumId": string(a.ID),
+ "userId": userID,
+ "subjectDesc": a.SubjectDesc,
+ })
+ if err != nil {
+ return "", fmt.Errorf("enqueue anchor job: %w", err)
+ }
+
+ if err := s.albums.UpdateAnchorJobID(ctx, id, userID, jobID); err != nil {
+ s.logger.Warn("failed to persist anchor job ID", "error", err, "album_id", string(id))
+ }
+
+ s.logger.Info("anchor generation enqueued", "album_id", string(id), "job_id", jobID)
+ return jobID, nil
+}
+
+// GenerateAllShots enqueues generation jobs for all pending shots.
+// Returns 422 if the album has no anchor yet (shots require an anchor reference).
+func (s *AlbumService) GenerateAllShots(ctx context.Context, id album.AlbumID, userID string) ([]string, error) {
+ a, err := s.albums.Get(ctx, id, userID)
+ if err != nil {
+ return nil, fmt.Errorf("album not found: %w", err)
+ }
+ if a.AnchorURL == "" {
+ return nil, album.ErrAnchorRequired
+ }
+
+ var jobIDs []string
+ for _, shot := range a.Shots {
+ if shot.Status != album.ShotPending && shot.Status != album.ShotFailed {
+ continue
+ }
+ jobID, err := s.enqueueShotJob(ctx, a, shot.Index)
+ if err != nil {
+ s.logger.Error("failed to enqueue shot", "error", err, "album_id", string(id), "shot_index", shot.Index)
+ continue
+ }
+ jobIDs = append(jobIDs, jobID)
+ }
+
+ s.logger.Info("shots enqueued", "album_id", string(id), "count", len(jobIDs))
+ return jobIDs, nil
+}
+
+// GenerateShot enqueues a generation job for a single shot (for regeneration).
+func (s *AlbumService) GenerateShot(ctx context.Context, id album.AlbumID, userID string, shotIndex int) (string, error) {
+ a, err := s.albums.Get(ctx, id, userID)
+ if err != nil {
+ return "", fmt.Errorf("album not found: %w", err)
+ }
+ if a.AnchorURL == "" {
+ return "", fmt.Errorf("anchor must be generated before shots")
+ }
+ if shotIndex < 0 || shotIndex >= len(a.Shots) {
+ return "", fmt.Errorf("shot index out of range: %d", shotIndex)
+ }
+ return s.enqueueShotJob(ctx, a, shotIndex)
+}
+
+// ResetShot clears a shot back to pending so it can be regenerated.
+func (s *AlbumService) ResetShot(ctx context.Context, id album.AlbumID, userID string, shotIndex int) error {
+ return s.albums.ResetShot(ctx, id, userID, shotIndex)
+}
+
+// enqueueShotJob is the internal helper that enqueues a single shot generation job.
+func (s *AlbumService) enqueueShotJob(ctx context.Context, a *album.Album, shotIndex int) (string, error) {
+ shot := a.Shots[shotIndex]
+ jobID, err := s.queue.Enqueue(ctx, "generate_shot", map[string]any{
+ "albumId": string(a.ID),
+ "userId": a.UserID,
+ "shotIndex": shotIndex,
+ "anchorUrl": a.AnchorURL,
+ "subjectDesc": a.SubjectDesc,
+ "direction": shot.Direction,
+ })
+ if err != nil {
+ return "", fmt.Errorf("enqueue shot job: %w", err)
+ }
+
+ if err := s.albums.UpdateShotJobID(ctx, a.ID, a.UserID, shotIndex, jobID); err != nil {
+ s.logger.Warn("failed to persist shot job ID", "error", err, "shot_index", shotIndex)
+ }
+
+ return jobID, nil
+}
+
diff --git a/services/persona-api/internal/service/auth.go b/services/persona-api/internal/service/auth.go
new file mode 100644
index 0000000..14c07e2
--- /dev/null
+++ b/services/persona-api/internal/service/auth.go
@@ -0,0 +1,644 @@
+package service
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "errors"
+ "fmt"
+ "math/big"
+ "net/url"
+ "strings"
+ "time"
+
+ "git.threesix.ai/jordan/persona-community-2/pkg/auth"
+ "git.threesix.ai/jordan/persona-community-2/pkg/logging"
+ "git.threesix.ai/jordan/persona-community-2/services/persona-api/internal/domain"
+ "git.threesix.ai/jordan/persona-community-2/services/persona-api/internal/port"
+)
+
+const (
+ // TokenLifetime is the access token duration (short-lived, requires refresh).
+ TokenLifetime = 15 * time.Minute
+ // SessionLifetime is how long a session stays valid before requiring re-login.
+ SessionLifetime = 30 * 24 * time.Hour // 30 days
+ // OTPExpiry is how long a one-time password is valid.
+ OTPExpiry = 10 * time.Minute
+ // MagicLinkExpiry is how long a magic link token is valid.
+ MagicLinkExpiry = 15 * time.Minute
+ // PasswordResetExpiry is how long a password reset token is valid.
+ PasswordResetExpiry = 1 * time.Hour
+ // EmailVerifyExpiry is how long an email verification code is valid.
+ EmailVerifyExpiry = 24 * time.Hour
+)
+
+// AuthService handles all authentication and identity flows.
+type AuthService struct {
+ users port.UserRepository
+ sessions port.SessionRepository
+ codes port.AuthCodeRepository
+ email port.EmailSender
+ jwtSecret []byte
+ issuer string
+ registrationEnabled bool
+ logger *logging.Logger
+}
+
+// NewAuthService creates a new auth service.
+func NewAuthService(
+ users port.UserRepository,
+ sessions port.SessionRepository,
+ codes port.AuthCodeRepository,
+ email port.EmailSender,
+ jwtSecret string,
+ registrationEnabled bool,
+ logger *logging.Logger,
+) *AuthService {
+ return &AuthService{
+ users: users,
+ sessions: sessions,
+ codes: codes,
+ email: email,
+ jwtSecret: []byte(jwtSecret),
+ issuer: "persona-community-2",
+ registrationEnabled: registrationEnabled,
+ logger: logger.WithService("AuthService"),
+ }
+}
+
+// LoginOutput is the result of a successful login or registration.
+type LoginOutput struct {
+ Token string
+ User *domain.User
+}
+
+// Register creates a new user account with email and password.
+func (s *AuthService) Register(ctx context.Context, email, password, name, ip, userAgent string) (*LoginOutput, error) {
+ if !s.registrationEnabled {
+ return nil, domain.ErrRegistrationDisabled
+ }
+
+ if err := auth.ValidatePasswordStrength(password); err != nil {
+ return nil, fmt.Errorf("%w: %w", domain.ErrWeakPassword, err)
+ }
+
+ name = strings.TrimSpace(name)
+ if len(name) > domain.MaxNameLen {
+ return nil, domain.ErrNameTooLong
+ }
+ if len(email) > domain.MaxEmailLen {
+ return nil, domain.ErrEmailTooLong
+ }
+
+ exists, err := s.users.ExistsByEmail(ctx, email)
+ if err != nil {
+ return nil, err
+ }
+ if exists {
+ return nil, domain.ErrDuplicateEmail
+ }
+
+ hash, err := auth.HashPassword(password)
+ if err != nil {
+ return nil, fmt.Errorf("hashing password: %w", err)
+ }
+
+ userID := domain.UserID("usr_" + generateID())
+ user := domain.NewUser(userID, email, name)
+
+ if err := s.users.Create(ctx, user); err != nil {
+ return nil, err
+ }
+ if err := s.users.SetPassword(ctx, userID, hash); err != nil {
+ return nil, err
+ }
+
+ s.logger.Info("user registered", "user_id", string(userID), "email", email)
+
+ return s.createSession(ctx, user, ip, userAgent)
+}
+
+// LoginWithPassword authenticates a user with email and password.
+func (s *AuthService) LoginWithPassword(ctx context.Context, email, password, ip, userAgent string) (*LoginOutput, error) {
+ user, err := s.users.GetByEmail(ctx, email)
+ if err != nil {
+ if errors.Is(err, domain.ErrUserNotFound) {
+ return nil, domain.ErrInvalidCredentials
+ }
+ return nil, err
+ }
+
+ if user.Status == domain.UserStatusSuspended {
+ return nil, domain.ErrUserSuspended
+ }
+
+ hash, err := s.users.GetPasswordHash(ctx, user.ID)
+ if err != nil {
+ return nil, err
+ }
+ if hash == "" || !auth.CheckPassword(password, hash) {
+ s.logger.Warn("invalid password attempt", "email", email)
+ return nil, domain.ErrInvalidCredentials
+ }
+
+ _ = s.users.UpdateLastLogin(ctx, user.ID)
+ s.logger.Info("user logged in", "user_id", string(user.ID), "email", email)
+
+ return s.createSession(ctx, user, ip, userAgent)
+}
+
+// RefreshToken issues a new access token if the session is still active.
+func (s *AuthService) RefreshToken(ctx context.Context, sessionID string, userID string) (*LoginOutput, error) {
+ sid := domain.SessionID(sessionID)
+ session, err := s.sessions.Get(ctx, sid)
+ if err != nil {
+ return nil, domain.ErrSessionNotFound
+ }
+ if !session.IsActive() {
+ return nil, domain.ErrSessionRevoked
+ }
+
+ user, err := s.users.Get(ctx, domain.UserID(userID))
+ if err != nil {
+ return nil, err
+ }
+ if user.Status == domain.UserStatusSuspended {
+ return nil, domain.ErrUserSuspended
+ }
+
+ _ = s.sessions.UpdateLastActive(ctx, sid)
+
+ token, err := s.generateToken(user, sessionID)
+ if err != nil {
+ return nil, err
+ }
+
+ return &LoginOutput{Token: token, User: user}, nil
+}
+
+// Logout revokes the current session.
+func (s *AuthService) Logout(ctx context.Context, sessionID string) error {
+ if sessionID == "" {
+ return nil
+ }
+ return s.sessions.Revoke(ctx, domain.SessionID(sessionID))
+}
+
+// LogoutAll revokes all sessions for a user, optionally keeping one.
+func (s *AuthService) LogoutAll(ctx context.Context, userID string, exceptSessionID *string) error {
+ var except *domain.SessionID
+ if exceptSessionID != nil {
+ sid := domain.SessionID(*exceptSessionID)
+ except = &sid
+ }
+ return s.sessions.RevokeAllForUser(ctx, domain.UserID(userID), except)
+}
+
+// CheckSession returns whether a session is active (not revoked, not expired).
+// Used as auth.SessionChecker for the SessionCheck middleware.
+func (s *AuthService) CheckSession(ctx context.Context, sessionID string) (bool, error) {
+ session, err := s.sessions.Get(ctx, domain.SessionID(sessionID))
+ if err != nil {
+ return false, nil
+ }
+ return session.IsActive(), nil
+}
+
+// ListSessions returns all active sessions for a user.
+func (s *AuthService) ListSessions(ctx context.Context, userID string) ([]domain.Session, error) {
+ return s.sessions.ListByUser(ctx, domain.UserID(userID))
+}
+
+// RevokeSession revokes a specific session for a user.
+func (s *AuthService) RevokeSession(ctx context.Context, userID, sessionID string) error {
+ session, err := s.sessions.Get(ctx, domain.SessionID(sessionID))
+ if err != nil {
+ return err
+ }
+ if session.UserID != domain.UserID(userID) {
+ return domain.ErrSessionNotFound
+ }
+ return s.sessions.Revoke(ctx, domain.SessionID(sessionID))
+}
+
+// GetCurrentUser returns the full user for the given ID.
+func (s *AuthService) GetCurrentUser(ctx context.Context, userID string) (*domain.User, error) {
+ return s.users.Get(ctx, domain.UserID(userID))
+}
+
+// UpdateProfile updates a user's name and avatar.
+func (s *AuthService) UpdateProfile(ctx context.Context, userID, name, avatarURL string) (*domain.User, error) {
+ user, err := s.users.Get(ctx, domain.UserID(userID))
+ if err != nil {
+ return nil, err
+ }
+
+ if name != "" {
+ name = strings.TrimSpace(name)
+ if len(name) > domain.MaxNameLen {
+ return nil, domain.ErrNameTooLong
+ }
+ user.Name = name
+ }
+ if avatarURL != "" {
+ if err := validateAvatarURL(avatarURL); err != nil {
+ return nil, err
+ }
+ user.AvatarURL = avatarURL
+ }
+
+ if err := s.users.Update(ctx, user); err != nil {
+ return nil, err
+ }
+ return user, nil
+}
+
+// ChangePassword changes a user's password after verifying the current one.
+func (s *AuthService) ChangePassword(ctx context.Context, userID, currentPassword, newPassword string) error {
+ uid := domain.UserID(userID)
+
+ hash, err := s.users.GetPasswordHash(ctx, uid)
+ if err != nil {
+ return err
+ }
+ if hash == "" || !auth.CheckPassword(currentPassword, hash) {
+ return domain.ErrInvalidCredentials
+ }
+
+ if err := auth.ValidatePasswordStrength(newPassword); err != nil {
+ return fmt.Errorf("%w: %w", domain.ErrWeakPassword, err)
+ }
+
+ newHash, err := auth.HashPassword(newPassword)
+ if err != nil {
+ return fmt.Errorf("hashing password: %w", err)
+ }
+
+ return s.users.SetPassword(ctx, uid, newHash)
+}
+
+// SendOTP generates a one-time password for the given email.
+// If the email is not registered and registration is enabled, the code is still
+// sent — the account will be created when the code is verified. This supports a
+// unified register+login flow with a single OTP email.
+func (s *AuthService) SendOTP(ctx context.Context, email, ip string) error {
+ user, err := s.users.GetByEmail(ctx, email)
+ if err != nil {
+ if !errors.Is(err, domain.ErrUserNotFound) {
+ return err
+ }
+ // Unknown email: only proceed if registration is open.
+ if !s.registrationEnabled {
+ s.logger.Info("OTP requested for unknown email (registration disabled)", "email", email)
+ return nil
+ }
+ // Registration enabled — send code anyway. UserID will be nil until verify.
+ user = nil
+ }
+
+ code := generateOTP()
+ var uid *domain.UserID
+ if user != nil {
+ uid = &user.ID
+ }
+ authCode := &domain.AuthCode{
+ ID: "acd_" + generateID(),
+ UserID: uid,
+ Email: email,
+ Code: code,
+ Purpose: domain.PurposeLoginOTP,
+ ExpiresAt: time.Now().Add(OTPExpiry),
+ IPAddress: ip,
+ CreatedAt: time.Now(),
+ }
+
+ if err := s.codes.Create(ctx, authCode); err != nil {
+ return err
+ }
+
+ s.logger.Info("auth code created", "purpose", "login_otp", "email", email, "code_id", authCode.ID)
+ if err := s.email.SendAuthCode(ctx, email, code, string(domain.PurposeLoginOTP)); err != nil {
+ s.logger.Error("failed to send OTP email", "email", email, "error", err)
+ }
+ return nil
+}
+
+// VerifyOTP verifies a one-time password and returns a login token.
+// If the email has no account yet and registration is enabled, the account is
+// created automatically — OTP delivery proves email ownership.
+func (s *AuthService) VerifyOTP(ctx context.Context, email, code, ip, userAgent string) (*LoginOutput, error) {
+ authCode, err := s.codes.FindValid(ctx, email, code, domain.PurposeLoginOTP)
+ if err != nil {
+ return nil, domain.ErrInvalidAuthCode
+ }
+
+ if err := s.codes.MarkUsed(ctx, authCode.ID); err != nil {
+ return nil, err
+ }
+
+ user, err := s.users.GetByEmail(ctx, email)
+ if err != nil {
+ if !errors.Is(err, domain.ErrUserNotFound) {
+ return nil, err
+ }
+ if !s.registrationEnabled {
+ return nil, domain.ErrRegistrationDisabled
+ }
+ // Auto-register: OTP delivery already proved email ownership.
+ user, err = s.autoRegisterViaOTP(ctx, email)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ _ = s.users.UpdateLastLogin(ctx, user.ID)
+ s.logger.Info("user logged in via OTP", "user_id", string(user.ID), "email", email)
+
+ return s.createSession(ctx, user, ip, userAgent)
+}
+
+// autoRegisterViaOTP creates a minimal user account for an email that just
+// verified an OTP. Email is considered verified because OTP delivery proved
+// ownership. The name defaults to the local part of the email address.
+func (s *AuthService) autoRegisterViaOTP(ctx context.Context, email string) (*domain.User, error) {
+ name := email
+ if at := strings.IndexByte(email, '@'); at > 0 {
+ name = email[:at]
+ }
+ userID := domain.UserID("usr_" + generateID())
+ user := domain.NewUser(userID, email, name)
+ user.EmailVerified = true // OTP delivery proves ownership
+
+ if err := s.users.Create(ctx, user); err != nil {
+ return nil, err
+ }
+ s.logger.Info("user auto-registered via OTP", "user_id", string(userID), "email", email)
+ return user, nil
+}
+
+// SendMagicLink generates and logs a magic link token.
+func (s *AuthService) SendMagicLink(ctx context.Context, email, ip string) error {
+ // Magic links can work for existing users.
+ // Don't reveal whether email exists — but propagate infrastructure errors.
+ user, err := s.users.GetByEmail(ctx, email)
+ if err != nil && !errors.Is(err, domain.ErrUserNotFound) {
+ return err
+ }
+
+ token := generateHexToken()
+ var uid *domain.UserID
+ if user != nil {
+ uid = &user.ID
+ }
+
+ authCode := &domain.AuthCode{
+ ID: "acd_" + generateID(),
+ UserID: uid,
+ Email: email,
+ Code: token,
+ Purpose: domain.PurposeMagicLink,
+ ExpiresAt: time.Now().Add(MagicLinkExpiry),
+ IPAddress: ip,
+ CreatedAt: time.Now(),
+ }
+
+ if err := s.codes.Create(ctx, authCode); err != nil {
+ return err
+ }
+
+ s.logger.Info("auth code created", "purpose", "magic_link", "email", email, "code_id", authCode.ID)
+ if err := s.email.SendAuthCode(ctx, email, token, string(domain.PurposeMagicLink)); err != nil {
+ s.logger.Error("failed to send magic link email", "email", email, "error", err)
+ }
+ return nil
+}
+
+// VerifyMagicLink verifies a magic link token and returns a login token.
+func (s *AuthService) VerifyMagicLink(ctx context.Context, email, token, ip, userAgent string) (*LoginOutput, error) {
+ authCode, err := s.codes.FindValid(ctx, email, token, domain.PurposeMagicLink)
+ if err != nil {
+ return nil, domain.ErrInvalidAuthCode
+ }
+
+ if err := s.codes.MarkUsed(ctx, authCode.ID); err != nil {
+ return nil, err
+ }
+
+ user, err := s.users.GetByEmail(ctx, email)
+ if err != nil {
+ return nil, err
+ }
+
+ _ = s.users.UpdateLastLogin(ctx, user.ID)
+ s.logger.Info("user logged in via magic link", "user_id", string(user.ID), "email", email)
+
+ return s.createSession(ctx, user, ip, userAgent)
+}
+
+// ForgotPassword generates a password reset token.
+func (s *AuthService) ForgotPassword(ctx context.Context, email, ip string) error {
+ user, err := s.users.GetByEmail(ctx, email)
+ if err != nil {
+ if errors.Is(err, domain.ErrUserNotFound) {
+ // Don't reveal whether email exists
+ s.logger.Info("password reset requested for unknown email", "email", email)
+ return nil
+ }
+ return err
+ }
+
+ token := generateHexToken()
+ uid := user.ID
+ authCode := &domain.AuthCode{
+ ID: "acd_" + generateID(),
+ UserID: &uid,
+ Email: email,
+ Code: token,
+ Purpose: domain.PurposePasswordReset,
+ ExpiresAt: time.Now().Add(PasswordResetExpiry),
+ IPAddress: ip,
+ CreatedAt: time.Now(),
+ }
+
+ if err := s.codes.Create(ctx, authCode); err != nil {
+ return err
+ }
+
+ s.logger.Info("auth code created", "purpose", "password_reset", "email", email, "code_id", authCode.ID)
+ if err := s.email.SendAuthCode(ctx, email, token, string(domain.PurposePasswordReset)); err != nil {
+ s.logger.Error("failed to send password reset email", "email", email, "error", err)
+ }
+ return nil
+}
+
+// ResetPassword sets a new password using a reset token and revokes all sessions.
+func (s *AuthService) ResetPassword(ctx context.Context, email, token, newPassword string) error {
+ authCode, err := s.codes.FindValid(ctx, email, token, domain.PurposePasswordReset)
+ if err != nil {
+ return domain.ErrInvalidAuthCode
+ }
+
+ if err := auth.ValidatePasswordStrength(newPassword); err != nil {
+ return fmt.Errorf("%w: %w", domain.ErrWeakPassword, err)
+ }
+
+ user, err := s.users.GetByEmail(ctx, email)
+ if err != nil {
+ return err
+ }
+
+ hash, err := auth.HashPassword(newPassword)
+ if err != nil {
+ return fmt.Errorf("hashing password: %w", err)
+ }
+
+ if err := s.users.SetPassword(ctx, user.ID, hash); err != nil {
+ return err
+ }
+ if err := s.codes.MarkUsed(ctx, authCode.ID); err != nil {
+ return err
+ }
+
+ // Revoke all sessions — user must re-login with new password.
+ _ = s.sessions.RevokeAllForUser(ctx, user.ID, nil)
+ s.logger.Info("password reset completed", "user_id", string(user.ID), "email", email)
+
+ return nil
+}
+
+// SendVerifyEmail generates an email verification code.
+func (s *AuthService) SendVerifyEmail(ctx context.Context, userID string) error {
+ user, err := s.users.Get(ctx, domain.UserID(userID))
+ if err != nil {
+ return err
+ }
+ if user.EmailVerified {
+ return nil
+ }
+
+ code := generateOTP()
+ uid := user.ID
+ authCode := &domain.AuthCode{
+ ID: "acd_" + generateID(),
+ UserID: &uid,
+ Email: user.Email,
+ Code: code,
+ Purpose: domain.PurposeEmailVerify,
+ ExpiresAt: time.Now().Add(EmailVerifyExpiry),
+ CreatedAt: time.Now(),
+ }
+
+ if err := s.codes.Create(ctx, authCode); err != nil {
+ return err
+ }
+
+ s.logger.Info("auth code created", "purpose", "email_verify", "email", user.Email, "code_id", authCode.ID)
+ if err := s.email.SendAuthCode(ctx, user.Email, code, string(domain.PurposeEmailVerify)); err != nil {
+ s.logger.Error("failed to send email verification", "email", user.Email, "error", err)
+ }
+ return nil
+}
+
+// VerifyEmail marks the user's email as verified.
+func (s *AuthService) VerifyEmail(ctx context.Context, userID, code string) error {
+ user, err := s.users.Get(ctx, domain.UserID(userID))
+ if err != nil {
+ return err
+ }
+
+ authCode, err := s.codes.FindValid(ctx, user.Email, code, domain.PurposeEmailVerify)
+ if err != nil {
+ return domain.ErrInvalidAuthCode
+ }
+
+ if err := s.codes.MarkUsed(ctx, authCode.ID); err != nil {
+ return err
+ }
+
+ user.EmailVerified = true
+ if err := s.users.Update(ctx, user); err != nil {
+ return err
+ }
+
+ s.logger.Info("email verified", "user_id", userID, "email", user.Email)
+ return nil
+}
+
+// createSession creates a session record and generates a JWT.
+func (s *AuthService) createSession(ctx context.Context, user *domain.User, ip, userAgent string) (*LoginOutput, error) {
+ sessionID := "ses_" + generateID()
+ now := time.Now()
+
+ session := &domain.Session{
+ ID: domain.SessionID(sessionID),
+ UserID: user.ID,
+ IPAddress: ip,
+ UserAgent: userAgent,
+ DeviceLabel: auth.ParseDeviceLabel(userAgent),
+ LastActiveAt: now,
+ ExpiresAt: now.Add(SessionLifetime),
+ CreatedAt: now,
+ }
+
+ if err := s.sessions.Create(ctx, session); err != nil {
+ return nil, err
+ }
+
+ token, err := s.generateToken(user, sessionID)
+ if err != nil {
+ return nil, err
+ }
+
+ return &LoginOutput{Token: token, User: user}, nil
+}
+
+// generateToken creates a JWT for the user with the given session ID.
+func (s *AuthService) generateToken(user *domain.User, sessionID string) (string, error) {
+ authUser := &auth.User{
+ ID: string(user.ID),
+ Email: user.Email,
+ Roles: user.Roles,
+ }
+ return auth.GenerateTokenWithSession(
+ s.jwtSecret, authUser, TokenLifetime, s.issuer, s.issuer, sessionID,
+ )
+}
+
+// generateID returns a random hex string suitable for entity IDs.
+func generateID() string {
+ b := make([]byte, 16)
+ if _, err := rand.Read(b); err != nil {
+ panic("crypto/rand failed: " + err.Error())
+ }
+ return hex.EncodeToString(b)
+}
+
+// generateOTP returns a 6-digit numeric one-time password.
+func generateOTP() string {
+ n, err := rand.Int(rand.Reader, big.NewInt(1000000))
+ if err != nil {
+ panic("crypto/rand failed: " + err.Error())
+ }
+ return fmt.Sprintf("%06d", n.Int64())
+}
+
+// validateAvatarURL checks that the URL uses http or https.
+func validateAvatarURL(rawURL string) error {
+ parsed, err := url.Parse(rawURL)
+ if err != nil {
+ return domain.ErrInvalidAvatarURL
+ }
+ if parsed.Scheme != "http" && parsed.Scheme != "https" {
+ return domain.ErrInvalidAvatarURL
+ }
+ return nil
+}
+
+// generateHexToken returns a 32-character hex token for magic links and resets.
+func generateHexToken() string {
+ b := make([]byte, 16)
+ if _, err := rand.Read(b); err != nil {
+ panic("crypto/rand failed: " + err.Error())
+ }
+ return hex.EncodeToString(b)
+}
diff --git a/services/persona-api/internal/service/example.go b/services/persona-api/internal/service/example.go
new file mode 100644
index 0000000..3a2e3bb
--- /dev/null
+++ b/services/persona-api/internal/service/example.go
@@ -0,0 +1,137 @@
+// Package service provides business logic / use cases for the application.
+// Services orchestrate domain operations using port interfaces.
+package service
+
+import (
+ "context"
+ "errors"
+
+ "github.com/google/uuid"
+
+ "git.threesix.ai/jordan/persona-community-2/pkg/logging"
+ "git.threesix.ai/jordan/persona-community-2/services/persona-api/internal/domain"
+ "git.threesix.ai/jordan/persona-community-2/services/persona-api/internal/port"
+)
+
+// ExampleService handles example-related business logic.
+type ExampleService struct {
+ repo port.ExampleRepository
+ logger *logging.Logger
+}
+
+// NewExampleService creates a new example service.
+func NewExampleService(repo port.ExampleRepository, logger *logging.Logger) *ExampleService {
+ return &ExampleService{
+ repo: repo,
+ logger: logger.WithService("ExampleService"),
+ }
+}
+
+// List returns all examples.
+func (s *ExampleService) List(ctx context.Context) ([]domain.Example, error) {
+ return s.repo.List(ctx)
+}
+
+// Get returns an example by ID.
+// Returns domain.ErrExampleNotFound if not found.
+func (s *ExampleService) Get(ctx context.Context, id domain.ExampleID) (*domain.Example, error) {
+ return s.repo.Get(ctx, id)
+}
+
+// CreateInput contains the data needed to create an example.
+type CreateInput struct {
+ Name string
+ Description string
+}
+
+// Create creates a new example with duplicate detection.
+// Returns domain.ErrDuplicateExample if name already exists.
+// Returns domain.ErrInvalidExampleName if name is invalid.
+func (s *ExampleService) Create(ctx context.Context, input CreateInput) (*domain.Example, error) {
+ // Check for duplicates
+ exists, err := s.repo.ExistsByName(ctx, input.Name)
+ if err != nil {
+ return nil, err
+ }
+ if exists {
+ return nil, domain.ErrDuplicateExample
+ }
+
+ // Generate new ID
+ id := domain.ExampleID(uuid.New().String())
+
+ // Create domain entity (validates name)
+ example, err := domain.NewExample(id, input.Name, input.Description)
+ if err != nil {
+ return nil, err
+ }
+
+ // Persist
+ if err := s.repo.Create(ctx, example); err != nil {
+ return nil, err
+ }
+
+ s.logger.Info("example created", "id", id, "name", input.Name)
+ return example, nil
+}
+
+// UpdateInput contains the data needed to update an example.
+type UpdateInput struct {
+ Name string
+ Description string
+}
+
+// Update modifies an existing example.
+// Returns domain.ErrExampleNotFound if not found.
+// Returns domain.ErrDuplicateExample if new name conflicts with another example.
+// Returns domain.ErrInvalidExampleName if name is invalid.
+func (s *ExampleService) Update(ctx context.Context, id domain.ExampleID, input UpdateInput) (*domain.Example, error) {
+ // Fetch existing
+ example, err := s.repo.Get(ctx, id)
+ if err != nil {
+ return nil, err
+ }
+
+ // Check for name conflicts (only if name changed)
+ if example.Name != input.Name {
+ exists, err := s.repo.ExistsByName(ctx, input.Name)
+ if err != nil {
+ return nil, err
+ }
+ if exists {
+ return nil, domain.ErrDuplicateExample
+ }
+ }
+
+ // Update domain entity (validates name)
+ if err := example.Update(input.Name, input.Description); err != nil {
+ return nil, err
+ }
+
+ // Persist
+ if err := s.repo.Update(ctx, example); err != nil {
+ return nil, err
+ }
+
+ s.logger.Info("example updated", "id", id, "name", input.Name)
+ return example, nil
+}
+
+// Delete removes an example by ID.
+// Returns domain.ErrExampleNotFound if not found.
+func (s *ExampleService) Delete(ctx context.Context, id domain.ExampleID) error {
+ // Verify exists before delete
+ if _, err := s.repo.Get(ctx, id); err != nil {
+ if errors.Is(err, domain.ErrExampleNotFound) {
+ return domain.ErrExampleNotFound
+ }
+ return err
+ }
+
+ if err := s.repo.Delete(ctx, id); err != nil {
+ return err
+ }
+
+ s.logger.Info("example deleted", "id", id)
+ return nil
+}
diff --git a/services/persona-api/internal/service/example_test.go b/services/persona-api/internal/service/example_test.go
new file mode 100644
index 0000000..117f869
--- /dev/null
+++ b/services/persona-api/internal/service/example_test.go
@@ -0,0 +1,282 @@
+package service
+
+import (
+ "context"
+ "sync"
+ "testing"
+
+ "git.threesix.ai/jordan/persona-community-2/pkg/logging"
+ "git.threesix.ai/jordan/persona-community-2/services/persona-api/internal/domain"
+ "git.threesix.ai/jordan/persona-community-2/services/persona-api/internal/port"
+)
+
+// mockExampleRepository implements port.ExampleRepository for testing.
+type mockExampleRepository struct {
+ mu sync.RWMutex
+ examples map[domain.ExampleID]*domain.Example
+}
+
+var _ port.ExampleRepository = (*mockExampleRepository)(nil)
+
+func newMockExampleRepository() *mockExampleRepository {
+ return &mockExampleRepository{
+ examples: make(map[domain.ExampleID]*domain.Example),
+ }
+}
+
+func (m *mockExampleRepository) List(ctx context.Context) ([]domain.Example, error) {
+ m.mu.RLock()
+ defer m.mu.RUnlock()
+
+ result := make([]domain.Example, 0, len(m.examples))
+ for _, e := range m.examples {
+ result = append(result, *e)
+ }
+ return result, nil
+}
+
+func (m *mockExampleRepository) Get(ctx context.Context, id domain.ExampleID) (*domain.Example, error) {
+ m.mu.RLock()
+ defer m.mu.RUnlock()
+
+ e, ok := m.examples[id]
+ if !ok {
+ return nil, domain.ErrExampleNotFound
+ }
+ // Return a copy to avoid mutation
+ copy := *e
+ return ©, nil
+}
+
+func (m *mockExampleRepository) Create(ctx context.Context, example *domain.Example) error {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ // Store a copy
+ copy := *example
+ m.examples[example.ID] = ©
+ return nil
+}
+
+func (m *mockExampleRepository) Update(ctx context.Context, example *domain.Example) error {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ if _, ok := m.examples[example.ID]; !ok {
+ return domain.ErrExampleNotFound
+ }
+ // Store a copy
+ copy := *example
+ m.examples[example.ID] = ©
+ return nil
+}
+
+func (m *mockExampleRepository) Delete(ctx context.Context, id domain.ExampleID) error {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ if _, ok := m.examples[id]; !ok {
+ return domain.ErrExampleNotFound
+ }
+ delete(m.examples, id)
+ return nil
+}
+
+func (m *mockExampleRepository) ExistsByName(ctx context.Context, name string) (bool, error) {
+ m.mu.RLock()
+ defer m.mu.RUnlock()
+
+ for _, e := range m.examples {
+ if e.Name == name {
+ return true, nil
+ }
+ }
+ return false, nil
+}
+
+func TestExampleService_Create(t *testing.T) {
+ repo := newMockExampleRepository()
+ svc := NewExampleService(repo, logging.Nop())
+
+ t.Run("creates example successfully", func(t *testing.T) {
+ example, err := svc.Create(context.Background(), CreateInput{
+ Name: "Test Example",
+ Description: "A test description",
+ })
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if example.Name != "Test Example" {
+ t.Errorf("expected name 'Test Example', got '%s'", example.Name)
+ }
+ if example.ID.IsZero() {
+ t.Error("expected non-empty ID")
+ }
+ })
+
+ t.Run("rejects duplicate name", func(t *testing.T) {
+ _, err := svc.Create(context.Background(), CreateInput{
+ Name: "Test Example",
+ Description: "Another description",
+ })
+ if err != domain.ErrDuplicateExample {
+ t.Errorf("expected ErrDuplicateExample, got %v", err)
+ }
+ })
+
+ t.Run("rejects empty name", func(t *testing.T) {
+ _, err := svc.Create(context.Background(), CreateInput{
+ Name: "",
+ Description: "Description",
+ })
+ if err != domain.ErrInvalidExampleName {
+ t.Errorf("expected ErrInvalidExampleName, got %v", err)
+ }
+ })
+}
+
+func TestExampleService_Get(t *testing.T) {
+ repo := newMockExampleRepository()
+ svc := NewExampleService(repo, logging.Nop())
+
+ // Create an example first
+ created, _ := svc.Create(context.Background(), CreateInput{
+ Name: "Get Test",
+ Description: "Description",
+ })
+
+ t.Run("returns existing example", func(t *testing.T) {
+ example, err := svc.Get(context.Background(), created.ID)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if example.Name != "Get Test" {
+ t.Errorf("expected name 'Get Test', got '%s'", example.Name)
+ }
+ })
+
+ t.Run("returns not found for missing example", func(t *testing.T) {
+ _, err := svc.Get(context.Background(), "nonexistent-id")
+ if err != domain.ErrExampleNotFound {
+ t.Errorf("expected ErrExampleNotFound, got %v", err)
+ }
+ })
+}
+
+func TestExampleService_Update(t *testing.T) {
+ repo := newMockExampleRepository()
+ svc := NewExampleService(repo, logging.Nop())
+
+ // Create examples
+ example1, _ := svc.Create(context.Background(), CreateInput{
+ Name: "Update Test 1",
+ Description: "Original",
+ })
+ _, _ = svc.Create(context.Background(), CreateInput{
+ Name: "Update Test 2",
+ Description: "Other",
+ })
+
+ t.Run("updates example successfully", func(t *testing.T) {
+ updated, err := svc.Update(context.Background(), example1.ID, UpdateInput{
+ Name: "Updated Name",
+ Description: "Updated description",
+ })
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if updated.Name != "Updated Name" {
+ t.Errorf("expected name 'Updated Name', got '%s'", updated.Name)
+ }
+ })
+
+ t.Run("allows same name on same example", func(t *testing.T) {
+ _, err := svc.Update(context.Background(), example1.ID, UpdateInput{
+ Name: "Updated Name",
+ Description: "Same name",
+ })
+ if err != nil {
+ t.Errorf("unexpected error updating with same name: %v", err)
+ }
+ })
+
+ t.Run("rejects name conflict", func(t *testing.T) {
+ _, err := svc.Update(context.Background(), example1.ID, UpdateInput{
+ Name: "Update Test 2",
+ Description: "Conflict",
+ })
+ if err != domain.ErrDuplicateExample {
+ t.Errorf("expected ErrDuplicateExample, got %v", err)
+ }
+ })
+
+ t.Run("returns not found for missing example", func(t *testing.T) {
+ _, err := svc.Update(context.Background(), "nonexistent-id", UpdateInput{
+ Name: "Anything",
+ Description: "",
+ })
+ if err != domain.ErrExampleNotFound {
+ t.Errorf("expected ErrExampleNotFound, got %v", err)
+ }
+ })
+}
+
+func TestExampleService_Delete(t *testing.T) {
+ repo := newMockExampleRepository()
+ svc := NewExampleService(repo, logging.Nop())
+
+ // Create an example first
+ created, _ := svc.Create(context.Background(), CreateInput{
+ Name: "Delete Test",
+ Description: "To be deleted",
+ })
+
+ t.Run("deletes example successfully", func(t *testing.T) {
+ err := svc.Delete(context.Background(), created.ID)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ // Verify deleted
+ _, err = svc.Get(context.Background(), created.ID)
+ if err != domain.ErrExampleNotFound {
+ t.Errorf("expected ErrExampleNotFound after delete, got %v", err)
+ }
+ })
+
+ t.Run("returns not found for missing example", func(t *testing.T) {
+ err := svc.Delete(context.Background(), "nonexistent-id")
+ if err != domain.ErrExampleNotFound {
+ t.Errorf("expected ErrExampleNotFound, got %v", err)
+ }
+ })
+}
+
+func TestExampleService_List(t *testing.T) {
+ repo := newMockExampleRepository()
+ svc := NewExampleService(repo, logging.Nop())
+
+ t.Run("returns empty list initially", func(t *testing.T) {
+ examples, err := svc.List(context.Background())
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if len(examples) != 0 {
+ t.Errorf("expected 0 examples, got %d", len(examples))
+ }
+ })
+
+ // Create some examples
+ _, _ = svc.Create(context.Background(), CreateInput{Name: "List Test 1", Description: ""})
+ _, _ = svc.Create(context.Background(), CreateInput{Name: "List Test 2", Description: ""})
+
+ t.Run("returns all examples", func(t *testing.T) {
+ examples, err := svc.List(context.Background())
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if len(examples) != 2 {
+ t.Errorf("expected 2 examples, got %d", len(examples))
+ }
+ })
+}
diff --git a/workers/media-worker/.env.example b/workers/media-worker/.env.example
new file mode 100644
index 0000000..3debabe
--- /dev/null
+++ b/workers/media-worker/.env.example
@@ -0,0 +1,25 @@
+# media-worker Worker Configuration
+
+# App
+APP_NAME=media-worker
+APP_ENVIRONMENT=development
+APP_DEBUG=true
+
+# Logging
+LOG_LEVEL=debug
+LOG_FORMAT=text
+
+# Database (required for job queue)
+# Local dev: PostgreSQL via docker-compose. Production: CockroachDB (platform-provisioned).
+# The postgres:// scheme works for both — CockroachDB is wire-compatible.
+DATABASE_URL=postgres://dev:dev@localhost:5432/persona-community-2?sslmode=disable
+
+# Worker
+WORKER_POLL_INTERVAL=10s
+WORKER_BATCH_SIZE=10
+WORKER_MAX_RETRIES=3
+WORKER_STALE_JOB_TIMEOUT=5m
+WORKER_JOB_TIMEOUT=5m
+
+# Redis (optional, for cache)
+# REDIS_URL=redis://localhost:6379/0
diff --git a/workers/media-worker/Dockerfile b/workers/media-worker/Dockerfile
new file mode 100644
index 0000000..bcc8c32
--- /dev/null
+++ b/workers/media-worker/Dockerfile
@@ -0,0 +1,33 @@
+# Build stage
+FROM golang:1.25-alpine AS builder
+
+RUN apk add --no-cache git
+
+# Configure Go private modules
+# Disable workspace mode - each component builds independently with replace directives
+ENV GOPRIVATE=git.threesix.ai/*
+ENV GOWORK=off
+
+WORKDIR /app
+
+# Copy shared pkg and this worker only
+COPY pkg/ ./pkg/
+COPY workers/media-worker/ ./workers/media-worker/
+
+# Download dependencies (populates go.sum if empty)
+RUN cd pkg && go mod download
+RUN cd workers/media-worker && go mod download
+
+# Build from the worker directory (uses replace directive for ../pkg)
+RUN cd workers/media-worker && CGO_ENABLED=0 go build -o /media-worker ./cmd/worker
+
+# Production stage
+FROM alpine:3.19
+
+RUN apk add --no-cache ca-certificates tzdata
+
+WORKDIR /
+
+COPY --from=builder /media-worker /media-worker
+
+ENTRYPOINT ["/media-worker"]
diff --git a/workers/media-worker/Makefile b/workers/media-worker/Makefile
new file mode 100644
index 0000000..a9d7d6a
--- /dev/null
+++ b/workers/media-worker/Makefile
@@ -0,0 +1,38 @@
+.PHONY: build run dev test lint fmt docker-build clean
+
+WORKER := media-worker
+BINARY := bin/$(WORKER)
+GO_MODULE := git.threesix.ai/jordan/persona-community-2
+
+# Build the worker binary
+build:
+ go build -o $(BINARY) ./cmd/worker
+
+# Run the worker locally
+run:
+ go run ./cmd/worker
+
+# Run the worker in development mode (alias for run)
+dev:
+ go run ./cmd/worker
+
+# Run tests
+test:
+ go test -v ./...
+
+# Run linter
+lint:
+ golangci-lint run ./...
+
+# Format code
+fmt:
+ gofmt -w .
+ goimports -w -local $(GO_MODULE) .
+
+# Build Docker image (run from monorepo root)
+docker-build:
+ docker build -t $(WORKER):latest -f Dockerfile ../..
+
+# Clean build artifacts
+clean:
+ rm -rf bin/
diff --git a/workers/media-worker/cmd/worker/main.go b/workers/media-worker/cmd/worker/main.go
new file mode 100644
index 0000000..ae158e1
--- /dev/null
+++ b/workers/media-worker/cmd/worker/main.go
@@ -0,0 +1,269 @@
+// Package main is the entry point for the media-worker worker.
+package main
+
+import (
+ "context"
+ "os"
+ "os/signal"
+ "syscall"
+ "time"
+
+ "github.com/redis/go-redis/v9"
+
+ "git.threesix.ai/jordan/persona-community-2/pkg/database"
+ "git.threesix.ai/jordan/persona-community-2/pkg/gemini"
+ "git.threesix.ai/jordan/persona-community-2/pkg/laozhang"
+ "git.threesix.ai/jordan/persona-community-2/pkg/logging"
+ "git.threesix.ai/jordan/persona-community-2/pkg/mediagen"
+ mediagenAdapters "git.threesix.ai/jordan/persona-community-2/pkg/mediagen/adapters"
+ "git.threesix.ai/jordan/persona-community-2/pkg/personagen"
+ "git.threesix.ai/jordan/persona-community-2/pkg/queue"
+ "git.threesix.ai/jordan/persona-community-2/pkg/realtime"
+ "git.threesix.ai/jordan/persona-community-2/pkg/storage"
+ "git.threesix.ai/jordan/persona-community-2/pkg/textgen"
+ textgenAdapters "git.threesix.ai/jordan/persona-community-2/pkg/textgen/adapters"
+ "git.threesix.ai/jordan/persona-community-2/workers/media-worker/internal/config"
+ "git.threesix.ai/jordan/persona-community-2/workers/media-worker/internal/handlers"
+)
+
+func main() {
+ // Initialize logger first (with defaults) so we can log config errors
+ logger := logging.New(logging.Config{
+ Level: logging.LevelInfo,
+ Format: logging.FormatJSON,
+ }).WithService("media-worker")
+
+ // Initialize configuration
+ cfg, err := config.Load()
+ if err != nil {
+ logger.Error("failed to load config", "error", err)
+ os.Exit(1)
+ }
+
+ // Reconfigure logger with loaded config
+ logger = logging.New(logging.Config{
+ Level: logging.ParseLevel(cfg.Logging.Level),
+ Format: logging.ParseFormat(cfg.Logging.Format),
+ Environment: cfg.AppConfig.Environment,
+ AddSource: cfg.AppConfig.IsDevelopment(),
+ }).WithService("media-worker")
+
+ logger.Info("starting media-worker worker")
+
+ // Setup graceful shutdown
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ // Connect to database
+ pool, err := database.Connect(ctx, cfg.Database.URL, database.Options{
+ MaxOpenConns: cfg.Database.MaxOpenConns,
+ MaxIdleConns: cfg.Database.MaxIdleConns,
+ ConnMaxLifetime: cfg.Database.ConnMaxLifetime,
+ })
+ if err != nil {
+ logger.Error("failed to connect to database", "error", err)
+ os.Exit(1)
+ }
+ defer pool.Close()
+ logger.Info("connected to database", "url", pool.URL)
+
+ // Run queue migrations (idempotent — safe for both service and worker)
+ if err := queue.RunMigrations(ctx, pool); err != nil {
+ logger.Error("failed to run queue migrations", "error", err)
+ os.Exit(1)
+ }
+ logger.Info("queue migrations complete")
+
+ // Initialize queue
+ jobQueue := queue.NewQueue(pool.DB, logger)
+
+ // Initialize Redis for SSE event publishing
+ if cfg.RedisURL == "" {
+ logger.Error("REDIS_URL is required for worker to publish SSE events")
+ os.Exit(1)
+ }
+ redisOpts, err := redis.ParseURL(cfg.RedisURL)
+ if err != nil {
+ logger.Error("failed to parse REDIS_URL", "error", err)
+ os.Exit(1)
+ }
+ redisClient := redis.NewClient(redisOpts)
+ if err := redisClient.Ping(ctx).Err(); err != nil {
+ logger.Error("failed to connect to Redis", "error", err)
+ os.Exit(1)
+ }
+ logger.Info("connected to Redis")
+
+ ssePub := realtime.NewSSEPublisher(redisClient, logger.Logger)
+
+ // Initialize AI providers
+ // LaoZhang client (primary provider — pay-per-use, OpenAI-compatible)
+ var laozhangClient *laozhang.Client
+ if apiKey := os.Getenv("LAOZHANG_API_KEY"); apiKey != "" {
+ laozhangClient, err = laozhang.NewClient(laozhang.Config{
+ APIKey: apiKey,
+ VideoTimeout: 5 * time.Minute,
+ Logger: logger.Logger,
+ })
+ if err != nil {
+ logger.Warn("failed to create LaoZhang client", "error", err)
+ } else {
+ logger.Info("LaoZhang client initialized")
+ }
+ }
+
+ // Gemini client for media generation
+ var geminiClient *gemini.Client
+ if apiKey := os.Getenv("GEMINI_API_KEY"); apiKey != "" {
+ geminiClient, err = gemini.NewClient(ctx, gemini.Config{
+ APIKey: apiKey,
+ Logger: logger.Logger,
+ })
+ if err != nil {
+ logger.Warn("failed to create Gemini client", "error", err)
+ } else {
+ logger.Info("Gemini client initialized")
+ }
+ }
+
+ // Create mediagen manager (image + video)
+ var mediagenManager *mediagen.Manager
+ {
+ var laozhangMediaProvider *mediagenAdapters.LaoZhangProvider
+ var geminiMediaProvider *mediagenAdapters.GeminiProvider
+ if laozhangClient != nil {
+ laozhangMediaProvider = mediagenAdapters.NewLaoZhangProvider(laozhangClient)
+ }
+ if geminiClient != nil {
+ geminiMediaProvider = mediagenAdapters.NewGeminiProvider(geminiClient)
+ }
+
+ if geminiMediaProvider != nil || laozhangMediaProvider != nil {
+ mgCfg := mediagen.ProductionConfig(mediagen.ProviderSet{
+ LaoZhang: laozhangMediaProvider,
+ Gemini: geminiMediaProvider,
+ }, mediagen.WithLogger(logger.Logger))
+ if laozhangMediaProvider != nil {
+ mgCfg.VideoProviders = append(mgCfg.VideoProviders, laozhangMediaProvider)
+ }
+ if geminiMediaProvider != nil {
+ mgCfg.VideoProviders = append(mgCfg.VideoProviders, geminiMediaProvider)
+ }
+ mediagenManager, err = mediagen.NewManager(mgCfg)
+ if err != nil {
+ logger.Warn("failed to create mediagen manager", "error", err)
+ } else {
+ logger.Info("mediagen manager initialized (image + video)")
+ }
+ }
+ }
+
+ // Create textgen manager (text + streaming)
+ var textgenManager *textgen.Manager
+ {
+ var textProviders []textgen.TextGenerator
+ if laozhangClient != nil {
+ textProviders = append(textProviders, textgenAdapters.NewLaoZhangTextProvider(laozhangClient, ""))
+ }
+ if apiKey := os.Getenv("GEMINI_API_KEY"); apiKey != "" {
+ geminiTextProvider, err := textgenAdapters.NewGeminiTextProvider(ctx, textgenAdapters.GeminiTextConfig{
+ APIKey: apiKey,
+ })
+ if err != nil {
+ logger.Warn("failed to create gemini text provider", "error", err)
+ } else {
+ textProviders = append(textProviders, geminiTextProvider)
+ }
+ }
+ if len(textProviders) > 0 {
+ textgenCfg := textgen.ProductionConfig(textgen.ProviderSet{}, textgen.WithLogger(logger.Logger))
+ textgenCfg.Providers = textProviders
+ textgenManager, err = textgen.NewManager(textgenCfg)
+ if err != nil {
+ logger.Warn("failed to create textgen manager", "error", err)
+ } else {
+ logger.Info("textgen manager initialized")
+ }
+ }
+ }
+
+ // Initialize and start handler
+ handler := handlers.New(logger, jobQueue, handlers.Config{
+ PollInterval: cfg.Worker.PollInterval,
+ StaleJobTimeout: cfg.Worker.StaleJobTimeout,
+ JobTimeout: cfg.Worker.JobTimeout,
+ })
+
+ // Initialize storage backend for persisting generated media.
+ // GCS_BUCKET is injected by the platform; if absent, store is nil (media not persisted).
+ var mediaStore storage.Store
+ if bucket := os.Getenv("GCS_BUCKET"); bucket != "" {
+ gcsStore, err := storage.NewGCSStore(ctx, bucket, os.Getenv("GCS_SERVICE_ACCOUNT_JSON"), logger.Logger)
+ if err != nil {
+ logger.Warn("failed to create GCS store, generated media will not be persisted", "error", err)
+ } else {
+ defer func() { _ = gcsStore.Close() }()
+ mediaStore = gcsStore
+ logger.Info("storage initialized (GCS)", "bucket", bucket)
+ }
+ }
+
+ // Register job handlers
+ if mediagenManager != nil {
+ handler.RegisterHandler("generate_image", handlers.ImageHandler(mediagenManager, mediaStore, ssePub, logger))
+ handler.RegisterHandler("generate_video", handlers.VideoHandler(mediagenManager, mediaStore, ssePub, logger))
+ }
+ if textgenManager != nil {
+ handler.RegisterHandler("generate_text", handlers.TextHandler(textgenManager, ssePub, logger))
+ handler.RegisterHandler("ai_chat_response", handlers.ChatResponseHandler(textgenManager, ssePub, logger))
+ }
+ // Persona generation requires both textgen (5-stage LLM pipeline) and mediagen (20 images + 4 videos).
+ if textgenManager != nil && mediagenManager != nil {
+ handler.RegisterHandler("persona_generate", personagen.QueueHandler(textgenManager, mediagenManager, mediaStore, ssePub, logger.Logger))
+ }
+
+ // Setup signal handling
+ sigCh := make(chan os.Signal, 1)
+ signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
+
+ // Start worker in goroutine
+ go handler.Run(ctx)
+
+ // Start stale job recovery in goroutine
+ go runStaleJobRecovery(ctx, jobQueue, cfg.Worker.StaleJobTimeout, logger)
+
+ // Wait for shutdown signal
+ sig := <-sigCh
+ logger.Info("received shutdown signal", "signal", sig.String())
+
+ // Trigger graceful shutdown with grace period
+ logger.Info("initiating graceful shutdown")
+ cancel()
+
+ // Give in-flight jobs time to complete (grace period)
+ const shutdownGracePeriod = 5 * time.Second
+ time.Sleep(shutdownGracePeriod)
+
+ logger.Info("media-worker worker stopped")
+}
+
+// runStaleJobRecovery periodically requeues jobs that have been running too long.
+func runStaleJobRecovery(ctx context.Context, q *queue.DBQueue, timeout time.Duration, logger *logging.Logger) {
+ const staleCheckInterval = time.Minute
+ ticker := time.NewTicker(staleCheckInterval)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ count, err := q.RequeueStale(ctx, timeout)
+ if err != nil {
+ logger.Error("failed to requeue stale jobs", "error", err)
+ } else if count > 0 {
+ logger.Info("requeued stale jobs", "count", count)
+ }
+ }
+ }
+}
diff --git a/workers/media-worker/cmd/worker/migrations/001_create_jobs.sql b/workers/media-worker/cmd/worker/migrations/001_create_jobs.sql
new file mode 100644
index 0000000..5af8ef9
--- /dev/null
+++ b/workers/media-worker/cmd/worker/migrations/001_create_jobs.sql
@@ -0,0 +1,32 @@
+-- Jobs queue table for async job processing.
+-- Used by pkg/queue for producer/consumer patterns.
+CREATE TABLE IF NOT EXISTS jobs (
+ id UUID PRIMARY KEY,
+ job_type VARCHAR(255) NOT NULL,
+ payload JSONB NOT NULL DEFAULT '{}',
+ status VARCHAR(50) NOT NULL DEFAULT 'pending',
+ priority INT NOT NULL DEFAULT 0,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+ started_at TIMESTAMPTZ,
+ completed_at TIMESTAMPTZ,
+ retry_count INT NOT NULL DEFAULT 0,
+ max_retries INT NOT NULL DEFAULT 3,
+ error TEXT,
+ worker_id VARCHAR(255)
+);
+
+-- Index for efficient dequeue: pending jobs ordered by priority (desc) and age (asc).
+-- Partial index only includes pending jobs for efficiency.
+CREATE INDEX IF NOT EXISTS idx_jobs_dequeue ON jobs (priority DESC, created_at ASC)
+ WHERE status = 'pending';
+
+-- Index for finding stale running jobs that need requeue.
+-- Used by RequeueStale to recover from crashed workers.
+CREATE INDEX IF NOT EXISTS idx_jobs_stale ON jobs (started_at)
+ WHERE status = 'running';
+
+-- Index for listing/filtering jobs by type.
+CREATE INDEX IF NOT EXISTS idx_jobs_type ON jobs (job_type, created_at DESC);
+
+-- Index for listing jobs by status (useful for monitoring dashboards).
+CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs (status, created_at DESC);
diff --git a/workers/media-worker/component.yaml b/workers/media-worker/component.yaml
new file mode 100644
index 0000000..0c326be
--- /dev/null
+++ b/workers/media-worker/component.yaml
@@ -0,0 +1,8 @@
+name: media-worker
+type: worker
+path: workers/media-worker
+dependencies: []
+# Add dependencies as needed:
+# - postgres
+# - redis
+# - rabbitmq
diff --git a/workers/media-worker/go.mod b/workers/media-worker/go.mod
new file mode 100644
index 0000000..23cd12d
--- /dev/null
+++ b/workers/media-worker/go.mod
@@ -0,0 +1,11 @@
+module git.threesix.ai/jordan/persona-community-2/workers/media-worker
+
+go 1.25
+
+require (
+ git.threesix.ai/jordan/persona-community-2/pkg v0.0.0
+ github.com/google/uuid v1.6.0
+)
+
+// Use local workspace modules (for Docker builds without go.work)
+replace git.threesix.ai/jordan/persona-community-2/pkg => ../../pkg
diff --git a/workers/media-worker/go.sum b/workers/media-worker/go.sum
new file mode 100644
index 0000000..e69de29
diff --git a/workers/media-worker/internal/config/config.go b/workers/media-worker/internal/config/config.go
new file mode 100644
index 0000000..29d4698
--- /dev/null
+++ b/workers/media-worker/internal/config/config.go
@@ -0,0 +1,71 @@
+// Package config provides worker-specific configuration.
+package config
+
+import (
+ "os"
+ "time"
+
+ "github.com/spf13/viper"
+
+ "git.threesix.ai/jordan/persona-community-2/pkg/config"
+)
+
+// Config holds media-worker worker configuration.
+type Config struct {
+ config.AppConfig
+ Database config.DatabaseConfig
+ Logging config.LoggingConfig
+ Worker WorkerConfig
+
+ // Redis for publishing SSE events to the service
+ RedisURL string
+}
+
+// WorkerConfig holds worker-specific settings.
+type WorkerConfig struct {
+ // PollInterval is how often to check for new jobs when queue is empty.
+ PollInterval time.Duration
+
+ // BatchSize is the max number of jobs to process per poll (for batch workers).
+ BatchSize int
+
+ // MaxRetries is the default maximum retry attempts for failed jobs.
+ MaxRetries int
+
+ // StaleJobTimeout is how long a job can run before being considered stale.
+ // Jobs running longer than this without heartbeat will be requeued.
+ StaleJobTimeout time.Duration
+
+ // JobTimeout is the maximum time a single job handler can run.
+ JobTimeout time.Duration
+}
+
+// Load reads configuration from environment variables.
+func Load() (*Config, error) {
+ if err := config.Init(config.Options{
+ AppName: "media-worker",
+ SetDefaults: func() {
+ viper.SetDefault("WORKER_POLL_INTERVAL", "10s")
+ viper.SetDefault("WORKER_BATCH_SIZE", 10)
+ viper.SetDefault("WORKER_MAX_RETRIES", 3)
+ viper.SetDefault("WORKER_STALE_JOB_TIMEOUT", "5m")
+ viper.SetDefault("WORKER_JOB_TIMEOUT", "5m")
+ },
+ }); err != nil {
+ return nil, err
+ }
+
+ return &Config{
+ AppConfig: config.ReadAppConfig(),
+ Database: config.ReadDatabaseConfig(),
+ Logging: config.ReadLoggingConfig(),
+ Worker: WorkerConfig{
+ PollInterval: viper.GetDuration("WORKER_POLL_INTERVAL"),
+ BatchSize: viper.GetInt("WORKER_BATCH_SIZE"),
+ MaxRetries: viper.GetInt("WORKER_MAX_RETRIES"),
+ StaleJobTimeout: viper.GetDuration("WORKER_STALE_JOB_TIMEOUT"),
+ JobTimeout: viper.GetDuration("WORKER_JOB_TIMEOUT"),
+ },
+ RedisURL: os.Getenv("REDIS_URL"),
+ }, nil
+}
diff --git a/workers/media-worker/internal/handlers/generate.go b/workers/media-worker/internal/handlers/generate.go
new file mode 100644
index 0000000..a4263b6
--- /dev/null
+++ b/workers/media-worker/internal/handlers/generate.go
@@ -0,0 +1,33 @@
+// Package handlers re-exports generation job handlers from the shared package.
+// The worker registers these handlers to process queue jobs.
+package handlers
+
+import (
+ "git.threesix.ai/jordan/persona-community-2/pkg/generation"
+ "git.threesix.ai/jordan/persona-community-2/pkg/logging"
+ "git.threesix.ai/jordan/persona-community-2/pkg/mediagen"
+ "git.threesix.ai/jordan/persona-community-2/pkg/queue"
+ "git.threesix.ai/jordan/persona-community-2/pkg/realtime"
+ "git.threesix.ai/jordan/persona-community-2/pkg/storage"
+ "git.threesix.ai/jordan/persona-community-2/pkg/textgen"
+)
+
+// ImageHandler returns a queue.Handler that processes image generation jobs.
+func ImageHandler(mg *mediagen.Manager, store storage.Store, pub realtime.EventPublisher, logger *logging.Logger) queue.Handler {
+ return generation.ImageHandler(mg, store, pub, logger)
+}
+
+// VideoHandler returns a queue.Handler that processes video generation jobs.
+func VideoHandler(mg *mediagen.Manager, store storage.Store, pub realtime.EventPublisher, logger *logging.Logger) queue.Handler {
+ return generation.VideoHandler(mg, store, pub, logger)
+}
+
+// TextHandler returns a queue.Handler that processes text generation jobs with streaming.
+func TextHandler(tg *textgen.Manager, pub realtime.EventPublisher, logger *logging.Logger) queue.Handler {
+ return generation.TextHandler(tg, pub, logger)
+}
+
+// ChatResponseHandler returns a queue.Handler that generates AI chat responses.
+func ChatResponseHandler(tg *textgen.Manager, pub realtime.EventPublisher, logger *logging.Logger) queue.Handler {
+ return generation.ChatResponseHandler(tg, pub, logger)
+}
diff --git a/workers/media-worker/internal/handlers/handler.go b/workers/media-worker/internal/handlers/handler.go
new file mode 100644
index 0000000..7d34012
--- /dev/null
+++ b/workers/media-worker/internal/handlers/handler.go
@@ -0,0 +1,147 @@
+// Package handlers provides the worker's job processing logic.
+package handlers
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "sync"
+ "time"
+
+ "github.com/google/uuid"
+
+ "git.threesix.ai/jordan/persona-community-2/pkg/logging"
+ "git.threesix.ai/jordan/persona-community-2/pkg/queue"
+)
+
+// Config holds handler configuration.
+type Config struct {
+ // PollInterval is how often to check for new jobs when queue is empty.
+ PollInterval time.Duration
+
+ // StaleJobTimeout is how long a job can run before being considered stale.
+ StaleJobTimeout time.Duration
+
+ // JobTimeout is the maximum time a job handler can run.
+ JobTimeout time.Duration
+}
+
+// Handler processes background jobs from the queue.
+type Handler struct {
+ logger *logging.Logger
+ queue queue.Consumer
+ handlers map[string]queue.Handler
+ config Config
+ workerID string
+ mu sync.RWMutex
+}
+
+// New creates a new Handler.
+func New(logger *logging.Logger, q queue.Consumer, cfg Config) *Handler {
+ // Apply defaults
+ if cfg.PollInterval == 0 {
+ cfg.PollInterval = 10 * time.Second
+ }
+ if cfg.StaleJobTimeout == 0 {
+ cfg.StaleJobTimeout = 5 * time.Minute
+ }
+ if cfg.JobTimeout == 0 {
+ cfg.JobTimeout = 5 * time.Minute
+ }
+
+ return &Handler{
+ logger: logger.WithComponent("handler"),
+ queue: q,
+ handlers: make(map[string]queue.Handler),
+ config: cfg,
+ workerID: uuid.New().String(),
+ }
+}
+
+// RegisterHandler registers a handler for a specific job type.
+// Call this before Run() to set up job processing.
+func (h *Handler) RegisterHandler(jobType string, handler queue.Handler) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ h.handlers[jobType] = handler
+ h.logger.Info("registered job handler", "type", jobType)
+}
+
+// Run starts the worker loop and processes jobs until context is cancelled.
+func (h *Handler) Run(ctx context.Context) {
+ h.logger.Info("worker loop started", "worker_id", h.workerID)
+
+ for {
+ select {
+ case <-ctx.Done():
+ h.logger.Info("worker loop stopping", "worker_id", h.workerID)
+ return
+ default:
+ if err := h.processNextJob(ctx); err != nil {
+ if errors.Is(err, queue.ErrNoJob) {
+ // Queue is empty, wait before polling again
+ select {
+ case <-ctx.Done():
+ return
+ case <-time.After(h.config.PollInterval):
+ continue
+ }
+ }
+ // Log error and continue
+ h.logger.Error("error processing job", "error", err)
+ time.Sleep(time.Second) // Brief pause on error
+ }
+ }
+ }
+}
+
+// processNextJob dequeues and processes a single job.
+func (h *Handler) processNextJob(ctx context.Context) error {
+ job, err := h.queue.Dequeue(ctx, h.workerID)
+ if err != nil {
+ return err
+ }
+
+ // Get handler for job type
+ h.mu.RLock()
+ handler, ok := h.handlers[job.Type]
+ h.mu.RUnlock()
+
+ if !ok {
+ h.logger.Error("no handler for job type", "job_id", job.ID, "type", job.Type)
+ return h.queue.Fail(ctx, job.ID, fmt.Sprintf("unknown job type: %s", job.Type))
+ }
+
+ // Apply middleware and process (TimeoutMiddleware handles the deadline)
+ wrappedHandler := queue.Chain(
+ queue.RecoveryMiddleware(h.logger),
+ queue.LoggingMiddleware(h.logger),
+ queue.TimeoutMiddleware(h.config.JobTimeout),
+ )(handler)
+
+ // Use parent context - TimeoutMiddleware applies the job timeout
+ jobCtx := ctx
+ _ = jobCtx // jobCtx used below
+
+ if err := wrappedHandler(jobCtx, job); err != nil {
+ // Truncate error message to prevent log bloat and potential data leakage
+ errMsg := truncateErrorMessage(err.Error(), 1000)
+ h.logger.Debug("job handler failed", "job_id", job.ID, "error", errMsg)
+ return h.queue.Fail(ctx, job.ID, errMsg)
+ }
+
+ return h.queue.Ack(ctx, job.ID)
+}
+
+// WorkerID returns this handler's unique worker identifier.
+func (h *Handler) WorkerID() string {
+ return h.workerID
+}
+
+// truncateErrorMessage limits error message length to prevent log bloat.
+func truncateErrorMessage(msg string, maxLen int) string {
+ if len(msg) <= maxLen {
+ return msg
+ }
+ return msg[:maxLen-3] + "..."
+}
+ {{- else}}
+ {{.AppName}}
+ {{- end}}
+