1package daemon23import (4 "context"5 "database/sql"6 "encoding/json"7 "errors"8 "fmt"9 "net/url"10 "strings"1112 "github.com/FelineStateMachine/tinyrelay/internal/catalog"13 "github.com/FelineStateMachine/tinyrelay/internal/community"14 "github.com/FelineStateMachine/tinyrelay/internal/replication"15 "github.com/FelineStateMachine/tinyrelay/internal/storage"16 "github.com/FelineStateMachine/tinyrelay/internal/templates"17)1819// AfterCreate finishes the durable work for a catalog row still in creating.20// It is intentionally separate from Catalog.Create so callers can retry after21// a crash without creating a second tenant or losing the assigned owner.22func (a *App) AfterCreate(ctx context.Context, meta catalog.Tenant, opts CreateOptions) error {23 template, ok := templates.Find(meta.Template)24 if !ok {25 return fmt.Errorf("creation: template %q not found", meta.Template)26 }27 source := strings.TrimSpace(opts.Source)28 if source == "" {29 source = meta.Source30 }31 if template.Source == "required" && source == "" {32 return fmt.Errorf("creation: template %q requires a source relay", template.Name)33 }34 if source != "" && !validSource(source) {35 return errors.New("creation: source must be an http, https, ws, or wss relay")36 }37 store, err := storage.Open(ctx, meta.Paths.Database)38 if err != nil {39 return fmt.Errorf("creation: open tenant store: %w", err)40 }41 defer store.Close()42 if err := initializeTenantRetryable(ctx, store, meta, template); err != nil {43 return err44 }45 if source == "" && template.Name == "inbox" {46 return seedInboxDiscoveryJob(ctx, store, meta, template)47 }48 if source == "" || !acceptsSource(template.Name) {49 return nil50 }51 return seedSourceJob(ctx, store, meta, template, source)52}5354func seedInboxDiscoveryJob(ctx context.Context, store *storage.Store, meta catalog.Tenant, template templates.Template) error {55 service, err := replication.NewService(replication.Config{Store: store, DataDir: meta.Paths.Root})56 if err != nil {57 return fmt.Errorf("creation: initialize replication: %w", err)58 }59 filterObject := map[string]any{"kinds": template.AllowKinds, "#p": []string{meta.Owner}}60 raw, err := json.Marshal(filterObject)61 if err != nil {62 return err63 }64 spec := replication.JobSpec{ID: "source-" + meta.ID, Kind: replication.JobPull, Filter: string(raw), Every: 1, DiscoverPubKey: meta.Owner}65 if err := service.AddJob(ctx, spec); err != nil {66 return fmt.Errorf("creation: seed inbox discovery job: %w", err)67 }68 return nil69}7071// RecoverCreating retries every interrupted creation and makes only tenants72// whose policy and source jobs were durably initialized ready.73func (a *App) RecoverCreating(ctx context.Context) error {74 items, err := a.catalog.List(ctx)75 if err != nil {76 return err77 }78 var failures []error79 for _, meta := range items {80 if meta.Status != catalog.StatusCreating {81 continue82 }83 if err := a.AfterCreate(ctx, meta, CreateOptions{Source: meta.Source, Owner: meta.Owner, Template: meta.Template}); err != nil {84 failures = append(failures, fmt.Errorf("%s: %w", meta.Name, err))85 continue86 }87 if err := a.catalog.MarkReady(ctx, meta.ID); err != nil {88 failures = append(failures, fmt.Errorf("%s: mark ready: %w", meta.Name, err))89 }90 }91 return errors.Join(failures...)92}9394func initializeTenantRetryable(ctx context.Context, store *storage.Store, meta catalog.Tenant, template templates.Template) error {95 p, err := templates.ApplyTemplate(template.Name, meta.Owner)96 if err != nil {97 return err98 }99 if _, err := community.New(ctx, store, p.Owner); err != nil {100 return err101 }102 err = store.WithTx(ctx, func(tx *sql.Tx) error {103 if err := storage.PutSetting(ctx, tx, "policy", p); err != nil {104 return err105 }106 for _, rule := range []struct {107 kinds []int108 name string109 }{{p.AllowedKinds, "allow"}, {p.BlockedKinds, "block"}} {110 for _, kind := range rule.kinds {111 if _, err := tx.ExecContext(ctx, "INSERT OR IGNORE INTO community_kind_rules(kind,rule) VALUES(?,?)", kind, rule.name); err != nil {112 return err113 }114 }115 }116 for _, retention := range p.Retention {117 if _, err := tx.ExecContext(ctx, "INSERT OR IGNORE INTO community_retention(kind,days) VALUES(?,?)", retention.Kind, retention.Days); err != nil {118 return err119 }120 }121 if _, err := tx.ExecContext(ctx, "CREATE TABLE IF NOT EXISTS config_connections(position INTEGER PRIMARY KEY,value TEXT NOT NULL)"); err != nil {122 return err123 }124 for index, connection := range template.Connections {125 raw, err := json.Marshal(map[string]string{"template": connection, "visibility": "public"})126 if err != nil {127 return err128 }129 if _, err := tx.ExecContext(ctx, "INSERT OR IGNORE INTO config_connections(position,value) VALUES(?,?)", index, string(raw)); err != nil {130 return err131 }132 }133 return nil134 })135 if err != nil {136 return fmt.Errorf("creation: initialize policy: %w", err)137 }138 return nil139}140141func seedSourceJob(ctx context.Context, store *storage.Store, meta catalog.Tenant, template templates.Template, source string) error {142 service, err := replication.NewService(replication.Config{Store: store, DataDir: meta.Paths.Root})143 if err != nil {144 return fmt.Errorf("creation: initialize replication: %w", err)145 }146 filter := ""147 if len(template.AllowKinds) > 0 {148 filterObject := map[string]any{"kinds": template.AllowKinds}149 if template.Name == "inbox" {150 filterObject["#p"] = []string{meta.Owner}151 }152 raw, err := json.Marshal(filterObject)153 if err != nil {154 return err155 }156 filter = string(raw)157 }158 kind := replication.JobPull159 every := template.EveryHours160 if template.Name == "inbox" {161 every = 1162 }163 if template.Name == "outbox" {164 kind = replication.JobMirror165 if every == 0 {166 every = 24167 }168 }169 spec := replication.JobSpec{ID: "source-" + meta.ID, Kind: kind, Relays: []string{sourceRelayURL(source)}, Filter: filter, Every: every}170 if err := service.AddJob(ctx, spec); err != nil {171 return fmt.Errorf("creation: seed source job: %w", err)172 }173 return nil174}175176func validSource(source string) bool {177 u, err := url.Parse(source)178 return err == nil && u.Host != "" && (u.Scheme == "http" || u.Scheme == "https" || u.Scheme == "ws" || u.Scheme == "wss") && u.User == nil179}180181func acceptsSource(name string) bool {182 return name == "search" || name == "articles" || name == "inbox" || name == "outbox"183}184185func sourceRelayURL(raw string) string {186 u, err := url.Parse(raw)187 if err != nil {188 return raw189 }190 switch u.Scheme {191 case "http":192 u.Scheme = "ws"193 case "https":194 u.Scheme = "wss"195 }196 return u.String()197}