| 1 | package daemon |
| 2 | |
| 3 | // Custom views hand fenced code blocks to an external transform and keep |
| 4 | // what comes back as relay-signed artifacts attached to the source event. |
| 5 | // This file holds the definition, its storage and the owner's management |
| 6 | // methods; custom_views_transform.go holds the block extraction, the POST |
| 7 | // and the artifacts, and custom_views_http.go serves them. |
| 8 | |
| 9 | import ( |
| 10 | "context" |
| 11 | "crypto/sha256" |
| 12 | "database/sql" |
| 13 | "encoding/hex" |
| 14 | "encoding/json" |
| 15 | "errors" |
| 16 | "fmt" |
| 17 | "net/http" |
| 18 | "sort" |
| 19 | "strconv" |
| 20 | "strings" |
| 21 | "sync" |
| 22 | "time" |
| 23 | |
| 24 | "github.com/FelineStateMachine/tinyrelay/internal/community" |
| 25 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 26 | "github.com/FelineStateMachine/tinyrelay/internal/gates" |
| 27 | "github.com/FelineStateMachine/tinyrelay/internal/records" |
| 28 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 29 | "github.com/FelineStateMachine/tinyrelay/internal/telemetry" |
| 30 | "github.com/FelineStateMachine/tinyrelay/internal/views" |
| 31 | gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit" |
| 32 | ) |
| 33 | |
| 34 | // customViewService owns custom view definitions, transform execution and |
| 35 | // artifact storage. It borrows tenant dependencies and serializes transforms |
| 36 | // per view. Git is bound after repository initialization and before workers |
| 37 | // start; Records supplies the signing identity for generated artifacts. |
| 38 | type customViewService struct { |
| 39 | store *storage.Store |
| 40 | community *community.Service |
| 41 | records *records.Service |
| 42 | git *gitrelay.GitRelay |
| 43 | telemetry *telemetry.Telemetry |
| 44 | tenantName string |
| 45 | publicURL string |
| 46 | loopback func() bool |
| 47 | resolveActor func(*http.Request) (string, error) |
| 48 | browseRead func(context.Context, string) error |
| 49 | gate *gates.Gate |
| 50 | viewClient *http.Client |
| 51 | viewMu sync.Mutex |
| 52 | viewIndex map[int][]customView |
| 53 | viewLocks map[string]*sync.Mutex |
| 54 | } |
| 55 | |
| 56 | type customViewServiceConfig struct { |
| 57 | Store *storage.Store |
| 58 | Community *community.Service |
| 59 | Records *records.Service |
| 60 | Git *gitrelay.GitRelay |
| 61 | Telemetry *telemetry.Telemetry |
| 62 | TenantName string |
| 63 | PublicURL string |
| 64 | Loopback func() bool |
| 65 | ResolveActor func(*http.Request) (string, error) |
| 66 | BrowseRead func(context.Context, string) error |
| 67 | Gate *gates.Gate |
| 68 | } |
| 69 | |
| 70 | func newCustomViewService(cfg customViewServiceConfig) *customViewService { |
| 71 | return &customViewService{ |
| 72 | store: cfg.Store, |
| 73 | community: cfg.Community, |
| 74 | records: cfg.Records, |
| 75 | git: cfg.Git, |
| 76 | telemetry: cfg.Telemetry, |
| 77 | tenantName: cfg.TenantName, |
| 78 | publicURL: cfg.PublicURL, |
| 79 | loopback: cfg.Loopback, |
| 80 | resolveActor: cfg.ResolveActor, |
| 81 | browseRead: cfg.BrowseRead, |
| 82 | gate: cfg.Gate, |
| 83 | } |
| 84 | } |
| 85 | |
| 86 | const ( |
| 87 | viewTransform = "view-transform" |
| 88 | viewKindMax = 16 |
| 89 | viewLanguageMax = 32 |
| 90 | viewLanguageLength = 32 |
| 91 | viewDefaultMaxBytes = 1 << 20 |
| 92 | viewMaxBytesCap = 4 << 20 |
| 93 | viewTimeout = 30 * time.Second |
| 94 | viewMaxAttempts = 3 |
| 95 | viewPauseFailures = 20 |
| 96 | viewBackfillLimit = 500 |
| 97 | viewHourlyPeriod = int64(3600) |
| 98 | viewResponseMaxBytes = int64(viewMaxBytesCap*4 + 1<<20) |
| 99 | ) |
| 100 | |
| 101 | // viewBackoff is the wait before the second and third attempt. |
| 102 | var viewBackoff = []time.Duration{time.Minute, 5 * time.Minute} |
| 103 | |
| 104 | const customViewSchema = `CREATE TABLE IF NOT EXISTS custom_views(name TEXT PRIMARY KEY, kinds TEXT NOT NULL, transform TEXT NOT NULL, trigger TEXT NOT NULL, audience TEXT NOT NULL, languages TEXT NOT NULL, max_bytes INTEGER NOT NULL, secret TEXT NOT NULL, generation TEXT NOT NULL DEFAULT '', enabled INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL, last_run_at INTEGER NOT NULL DEFAULT 0, last_status TEXT NOT NULL DEFAULT '', failures INTEGER NOT NULL DEFAULT 0); |
| 105 | CREATE TABLE IF NOT EXISTS custom_view_artifacts(view TEXT NOT NULL, hash TEXT NOT NULL, event_id TEXT NOT NULL, type TEXT NOT NULL, engine TEXT NOT NULL DEFAULT '' , audience TEXT NOT NULL, content TEXT NOT NULL, raw TEXT NOT NULL, created_at INTEGER NOT NULL, PRIMARY KEY(view,hash)); |
| 106 | CREATE TABLE IF NOT EXISTS custom_view_sources(view TEXT NOT NULL, hash TEXT NOT NULL, source TEXT NOT NULL, block INTEGER NOT NULL, expires INTEGER NOT NULL DEFAULT 0, PRIMARY KEY(view,hash,source)); |
| 107 | CREATE INDEX IF NOT EXISTS custom_view_sources_source ON custom_view_sources(view,source);` |
| 108 | |
| 109 | // customView is one stored definition. The secret never leaves the tenant |
| 110 | // except once, in the addcustomview answer. |
| 111 | type customView struct { |
| 112 | Name string `json:"name"` |
| 113 | Kinds []int `json:"kinds"` |
| 114 | Transform string `json:"transform"` |
| 115 | Trigger string `json:"trigger"` |
| 116 | Audience string `json:"audience"` |
| 117 | Languages []string `json:"languages"` |
| 118 | MaxBytes int `json:"maxBytes"` |
| 119 | Enabled bool `json:"enabled"` |
| 120 | CreatedAt int64 `json:"created"` |
| 121 | LastRunAt int64 `json:"lastRun"` |
| 122 | LastStatus string `json:"lastStatus"` |
| 123 | Failures int `json:"failures"` |
| 124 | secret string |
| 125 | generation string |
| 126 | } |
| 127 | |
| 128 | type customViewTarget struct { |
| 129 | Name string |
| 130 | Fingerprint string |
| 131 | } |
| 132 | |
| 133 | func customViewFingerprint(view customView) string { |
| 134 | canonical := struct { |
| 135 | Name string |
| 136 | Kinds []int |
| 137 | Transform string |
| 138 | Trigger string |
| 139 | Audience string |
| 140 | Languages []string |
| 141 | MaxBytes int |
| 142 | Secret string |
| 143 | CreatedAt int64 |
| 144 | Generation string |
| 145 | }{view.Name, view.Kinds, view.Transform, view.Trigger, view.Audience, view.Languages, view.MaxBytes, view.secret, view.CreatedAt, view.generation} |
| 146 | raw, _ := json.Marshal(canonical) |
| 147 | sum := sha256.Sum256(raw) |
| 148 | return hex.EncodeToString(sum[:]) |
| 149 | } |
| 150 | |
| 151 | func (v customView) summary() map[string]any { |
| 152 | state := "active" |
| 153 | if !v.Enabled { |
| 154 | state = "paused" |
| 155 | } |
| 156 | return map[string]any{"name" : v.Name, "kinds" : v.Kinds, "transform" : v.Transform, "host" : callbackHost(v.Transform), "trigger" : v.Trigger, "audience" : v.Audience, "languages" : v.Languages, "maxBytes" : v.MaxBytes, "enabled" : v.Enabled, "state" : state, "created" : v.CreatedAt, "lastRun" : v.LastRunAt, "lastStatus" : v.LastStatus, "failures" : v.Failures} |
| 157 | } |
| 158 | |
| 159 | func (v customView) watches(kind int) bool { |
| 160 | for _, k := range v.Kinds { |
| 161 | if k == kind { |
| 162 | return true |
| 163 | } |
| 164 | } |
| 165 | return false |
| 166 | } |
| 167 | |
| 168 | func (t *customViewService) initCustomViews(ctx context.Context) error { |
| 169 | if _, err := t.store.DB().ExecContext(ctx, customViewSchema); err != nil { |
| 170 | return err |
| 171 | } |
| 172 | _, err := t.store.DB().ExecContext(ctx, `ALTER TABLE custom_views ADD COLUMN generation TEXT NOT NULL DEFAULT ''` ) |
| 173 | if err != nil && !strings.Contains(strings.ToLower(err.Error()), "duplicate column" ) { |
| 174 | return err |
| 175 | } |
| 176 | rows, err := t.store.DB().QueryContext(ctx, `SELECT name FROM custom_views WHERE generation=''` ) |
| 177 | if err != nil { |
| 178 | return err |
| 179 | } |
| 180 | defer rows.Close() |
| 181 | var names []string |
| 182 | for rows.Next() { |
| 183 | var name string |
| 184 | if err := rows.Scan(&name); err != nil { |
| 185 | return err |
| 186 | } |
| 187 | names = append(names, name) |
| 188 | } |
| 189 | if err := rows.Err(); err != nil { |
| 190 | return err |
| 191 | } |
| 192 | for _, name := range names { |
| 193 | generation, err := randomHex(16) |
| 194 | if err != nil { |
| 195 | return err |
| 196 | } |
| 197 | if _, err := t.store.DB().ExecContext(ctx, `UPDATE custom_views SET generation=? WHERE name=? AND generation=''` , generation, name); err != nil { |
| 198 | return err |
| 199 | } |
| 200 | } |
| 201 | return nil |
| 202 | } |
| 203 | |
| 204 | func customViewMethod(method string) bool { |
| 205 | return containsString([]string{"listcustomviews" , "addcustomview" , "removecustomview" , "runcustomview" , "pausecustomview" , "resumecustomview" }, method) |
| 206 | } |
| 207 | |
| 208 | // customViewExecute serves the owner's custom view methods under the view |
| 209 | // operation. Every change is recorded in the audit log. |
| 210 | func (t *customViewService) customViewExecute(ctx context.Context, actor, method string, params []json.RawMessage) (result any, err error) { |
| 211 | ctx, finish := t.telemetry.Start(ctx, "view" ) |
| 212 | outcome := "error" |
| 213 | defer func() { |
| 214 | if err != nil { |
| 215 | outcome = callbackOutcome(err) |
| 216 | } |
| 217 | finish(outcome) |
| 218 | }() |
| 219 | role, err := t.community.Role(ctx, actor) |
| 220 | if err != nil { |
| 221 | return nil, err |
| 222 | } |
| 223 | if role != "owner" { |
| 224 | return nil, fmt.Errorf("restricted: only the owner may %s" , method) |
| 225 | } |
| 226 | switch method { |
| 227 | case "listcustomviews" : |
| 228 | rows, err := t.customViewRows(ctx) |
| 229 | if err != nil { |
| 230 | return nil, err |
| 231 | } |
| 232 | out := make([]map[string]any, 0, len(rows)) |
| 233 | for _, row := range rows { |
| 234 | out = append(out, row.summary()) |
| 235 | } |
| 236 | outcome = "ok" |
| 237 | return out, nil |
| 238 | case "addcustomview" : |
| 239 | result, err = t.addCustomView(ctx, actor, params) |
| 240 | default: |
| 241 | result, err = t.changeCustomView(ctx, actor, method, params) |
| 242 | } |
| 243 | if err == nil { |
| 244 | outcome = "ok" |
| 245 | } |
| 246 | return result, err |
| 247 | } |
| 248 | |
| 249 | // customViewOptions is the addcustomview parameter. Kinds and languages may |
| 250 | // arrive as lists or as comma-separated strings from a plain form. |
| 251 | type customViewOptions struct { |
| 252 | Name string `json:"name"` |
| 253 | Kinds json.RawMessage `json:"kinds"` |
| 254 | Transform string `json:"transform"` |
| 255 | Trigger string `json:"trigger"` |
| 256 | Audience string `json:"audience"` |
| 257 | Languages json.RawMessage `json:"languages"` |
| 258 | MaxBytes json.RawMessage `json:"max_bytes"` |
| 259 | Secret string `json:"secret"` |
| 260 | } |
| 261 | |
| 262 | func (t *customViewService) addCustomView(ctx context.Context, actor string, params []json.RawMessage) (any, error) { |
| 263 | if len(params) != 1 { |
| 264 | return nil, errors.New("invalid: addcustomview expects one object with name, kinds, transform, trigger, audience, languages and optional max_bytes and secret" ) |
| 265 | } |
| 266 | var options customViewOptions |
| 267 | if err := json.Unmarshal(params[0], &options); err != nil { |
| 268 | return nil, errors.New("invalid: addcustomview expects one object with name, kinds, transform, trigger, audience, languages and optional max_bytes and secret" ) |
| 269 | } |
| 270 | view, err := t.parseCustomView(options) |
| 271 | if err != nil { |
| 272 | return nil, err |
| 273 | } |
| 274 | generated := false |
| 275 | secret := strings.TrimSpace(options.Secret) |
| 276 | if secret == "" { |
| 277 | secret, err = randomHex(32) |
| 278 | if err != nil { |
| 279 | return nil, err |
| 280 | } |
| 281 | generated = true |
| 282 | } else if len(secret) < callbackSecretMin || len(secret) > callbackSecretMax || strings.Trim(secret, "!\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~" ) != "" { |
| 283 | return nil, fmt.Errorf("invalid: secret must be %d to %d printable ASCII characters" , callbackSecretMin, callbackSecretMax) |
| 284 | } |
| 285 | view.secret = secret |
| 286 | view.generation, err = randomHex(16) |
| 287 | if err != nil { |
| 288 | return nil, err |
| 289 | } |
| 290 | view.Enabled = true |
| 291 | view.CreatedAt = time.Now().Unix() |
| 292 | kinds, _ := json.Marshal(view.Kinds) |
| 293 | languages, _ := json.Marshal(view.Languages) |
| 294 | _, err = t.store.DB().ExecContext(ctx, `INSERT INTO custom_views(name,kinds,transform,trigger,audience,languages,max_bytes,secret,generation,enabled,created_at) VALUES(?,?,?,?,?,?,?,?,?,1,?)` , view.Name, string(kinds), view.Transform, view.Trigger, view.Audience, string(languages), view.MaxBytes, secret, view.generation, view.CreatedAt) |
| 295 | if err != nil { |
| 296 | if strings.Contains(err.Error(), "UNIQUE" ) { |
| 297 | return nil, errors.New("invalid: a view with that name exists" ) |
| 298 | } |
| 299 | return nil, err |
| 300 | } |
| 301 | t.invalidateCustomViews() |
| 302 | if err := t.community.Record(ctx, actor, "addcustomview" , view.Name, "" ); err != nil { |
| 303 | return nil, err |
| 304 | } |
| 305 | // Counts only: the transform host, path and secret stay out of the log. |
| 306 | t.telemetry.Logger().Info("custom view added" , "tenant" , t.tenantName, "kinds" , len(view.Kinds), "languages" , len(view.Languages), "generated" , generated) |
| 307 | out := view.summary() |
| 308 | out["secret" ] = secret |
| 309 | return out, nil |
| 310 | } |
| 311 | |
| 312 | // parseCustomView validates a definition: the name, the kinds it watches, |
| 313 | // the https transform, the trigger, the audience, the languages and the |
| 314 | // artifact size limit. |
| 315 | func (t *customViewService) parseCustomView(options customViewOptions) (customView, error) { |
| 316 | view := customView{Name: strings.TrimSpace(options.Name), Transform: strings.TrimSpace(options.Transform), Trigger: strings.TrimSpace(options.Trigger), Audience: strings.TrimSpace(options.Audience), MaxBytes: viewDefaultMaxBytes} |
| 317 | if !views.NamePattern.MatchString(view.Name) { |
| 318 | return customView{}, errors.New("invalid: name must be 1 to 32 lowercase letters, digits or hyphens" ) |
| 319 | } |
| 320 | kinds, err := intList(options.Kinds) |
| 321 | if err != nil || len(kinds) == 0 || len(kinds) > viewKindMax { |
| 322 | return customView{}, fmt.Errorf("invalid: kinds must list 1 to %d event kinds" , viewKindMax) |
| 323 | } |
| 324 | for _, kind := range kinds { |
| 325 | if kind < 0 || kind > 65535 { |
| 326 | return customView{}, errors.New("invalid: kinds must be between 0 and 65535" ) |
| 327 | } |
| 328 | } |
| 329 | sort.Ints(kinds) |
| 330 | view.Kinds = dedupeInts(kinds) |
| 331 | if err := validateWebhookURL(view.Transform, t.publicURL); err != nil { |
| 332 | return customView{}, fmt.Errorf("invalid: transform %s" , strings.TrimPrefix(err.Error(), "invalid: " )) |
| 333 | } |
| 334 | if view.Trigger == "" { |
| 335 | view.Trigger = "write" |
| 336 | } |
| 337 | if view.Trigger != "write" && view.Trigger != "hourly" { |
| 338 | return customView{}, errors.New("invalid: trigger must be write or hourly" ) |
| 339 | } |
| 340 | if view.Audience == "" { |
| 341 | view.Audience = "public" |
| 342 | } |
| 343 | if view.Audience != "public" && view.Audience != "members" { |
| 344 | return customView{}, errors.New("invalid: audience must be public or members" ) |
| 345 | } |
| 346 | languages, err := stringList(options.Languages) |
| 347 | if err != nil || len(languages) == 0 || len(languages) > viewLanguageMax { |
| 348 | return customView{}, fmt.Errorf("invalid: languages must list 1 to %d fenced block languages" , viewLanguageMax) |
| 349 | } |
| 350 | seen := map[string]bool{} |
| 351 | view.Languages = nil |
| 352 | for _, lang := range languages { |
| 353 | lang = strings.ToLower(strings.TrimSpace(lang)) |
| 354 | if lang == "" || len(lang) > viewLanguageLength || strings.Trim(lang, "abcdefghijklmnopqrstuvwxyz0123456789+-_.#" ) != "" { |
| 355 | return customView{}, errors.New("invalid: languages must be lowercase words of letters, digits, +, -, _, . or #" ) |
| 356 | } |
| 357 | if !seen[lang] { |
| 358 | seen[lang] = true |
| 359 | view.Languages = append(view.Languages, lang) |
| 360 | } |
| 361 | } |
| 362 | if len(options.MaxBytes) > 0 && string(options.MaxBytes) != "null" && string(options.MaxBytes) != `""` { |
| 363 | var size int |
| 364 | var text string |
| 365 | if json.Unmarshal(options.MaxBytes, &size) != nil { |
| 366 | if json.Unmarshal(options.MaxBytes, &text) != nil { |
| 367 | return customView{}, errors.New("invalid: max_bytes must be a number" ) |
| 368 | } |
| 369 | if size, err = strconv.Atoi(strings.TrimSpace(text)); err != nil { |
| 370 | return customView{}, errors.New("invalid: max_bytes must be a number" ) |
| 371 | } |
| 372 | } |
| 373 | if size < 1 || size > viewMaxBytesCap { |
| 374 | return customView{}, fmt.Errorf("invalid: max_bytes must be between 1 and %d" , viewMaxBytesCap) |
| 375 | } |
| 376 | view.MaxBytes = size |
| 377 | } |
| 378 | return view, nil |
| 379 | } |
| 380 | |
| 381 | // intList reads a JSON list of integers, a list of numeric strings or a |
| 382 | // comma-separated string. |
| 383 | func intList(raw json.RawMessage) ([]int, error) { |
| 384 | if len(raw) == 0 { |
| 385 | return nil, nil |
| 386 | } |
| 387 | var out []int |
| 388 | if json.Unmarshal(raw, &out) == nil { |
| 389 | return out, nil |
| 390 | } |
| 391 | items, err := stringList(raw) |
| 392 | if err != nil { |
| 393 | return nil, err |
| 394 | } |
| 395 | for _, item := range items { |
| 396 | value, err := strconv.Atoi(strings.TrimSpace(item)) |
| 397 | if err != nil { |
| 398 | return nil, err |
| 399 | } |
| 400 | out = append(out, value) |
| 401 | } |
| 402 | return out, nil |
| 403 | } |
| 404 | |
| 405 | // stringList reads a JSON list of strings or a comma-separated string. |
| 406 | func stringList(raw json.RawMessage) ([]string, error) { |
| 407 | if len(raw) == 0 { |
| 408 | return nil, nil |
| 409 | } |
| 410 | var out []string |
| 411 | if json.Unmarshal(raw, &out) == nil { |
| 412 | return out, nil |
| 413 | } |
| 414 | var text string |
| 415 | if err := json.Unmarshal(raw, &text); err != nil { |
| 416 | return nil, err |
| 417 | } |
| 418 | for _, item := range strings.FieldsFunc(text, func(r rune) bool { return r == ',' || r == ' ' || r == '\n' }) { |
| 419 | if item = strings.TrimSpace(item); item != "" { |
| 420 | out = append(out, item) |
| 421 | } |
| 422 | } |
| 423 | return out, nil |
| 424 | } |
| 425 | |
| 426 | func dedupeInts(values []int) []int { |
| 427 | out := values[:0] |
| 428 | for i, value := range values { |
| 429 | if i == 0 || value != values[i-1] { |
| 430 | out = append(out, value) |
| 431 | } |
| 432 | } |
| 433 | return out |
| 434 | } |
| 435 | |
| 436 | func (t *customViewService) changeCustomView(ctx context.Context, actor, method string, params []json.RawMessage) (any, error) { |
| 437 | name := customViewNameParam(params) |
| 438 | if name == "" { |
| 439 | return nil, errors.New("invalid: view name required" ) |
| 440 | } |
| 441 | view, err := t.customViewByName(ctx, name) |
| 442 | if err != nil { |
| 443 | return nil, err |
| 444 | } |
| 445 | switch method { |
| 446 | case "removecustomview" : |
| 447 | if err := t.removeCustomView(ctx, view); err != nil { |
| 448 | return nil, err |
| 449 | } |
| 450 | case "pausecustomview" : |
| 451 | _, err = t.store.DB().ExecContext(ctx, `UPDATE custom_views SET enabled=0, last_status='paused' WHERE name=?` , name) |
| 452 | view.Enabled, view.LastStatus = false, "paused" |
| 453 | case "resumecustomview" : |
| 454 | _, err = t.store.DB().ExecContext(ctx, `UPDATE custom_views SET enabled=1, failures=0, last_status='' WHERE name=?` , name) |
| 455 | view.Enabled, view.Failures, view.LastStatus = true, 0, "" |
| 456 | case "runcustomview" : |
| 457 | if !view.Enabled { |
| 458 | return nil, errors.New("invalid: resume the view before running it" ) |
| 459 | } |
| 460 | refresh := customViewRefreshParam(params) |
| 461 | queued, err := t.queueCustomViewBackfill(ctx, view, 0, strconv.FormatInt(time.Now().UnixNano(), 10), refresh) |
| 462 | if err != nil { |
| 463 | return nil, err |
| 464 | } |
| 465 | if err := t.community.Record(ctx, actor, method, name, strconv.Itoa(queued)); err != nil { |
| 466 | return nil, err |
| 467 | } |
| 468 | t.telemetry.Logger().Info("custom view run queued" , "tenant" , t.tenantName, "queued" , queued) |
| 469 | out := view.summary() |
| 470 | out["queued" ] = queued |
| 471 | return out, nil |
| 472 | default: |
| 473 | return nil, fmt.Errorf("unsupported: management method %q" , method) |
| 474 | } |
| 475 | if err != nil { |
| 476 | return nil, err |
| 477 | } |
| 478 | t.invalidateCustomViews() |
| 479 | if err := t.community.Record(ctx, actor, method, name, "" ); err != nil { |
| 480 | return nil, err |
| 481 | } |
| 482 | if method == "removecustomview" { |
| 483 | return map[string]any{"name" : name, "removed" : true}, nil |
| 484 | } |
| 485 | return view.summary(), nil |
| 486 | } |
| 487 | |
| 488 | // removeCustomView deletes the definition and every artifact it produced, |
| 489 | // including the stored public records. |
| 490 | func (t *customViewService) removeCustomView(ctx context.Context, view customView) error { |
| 491 | ids, err := t.customViewArtifactIDs(ctx, `SELECT event_id FROM custom_view_artifacts WHERE view=?` , view.Name) |
| 492 | if err != nil { |
| 493 | return err |
| 494 | } |
| 495 | if err := t.store.WithTx(ctx, func(tx *sql.Tx) error { |
| 496 | for _, table := range []string{"custom_view_sources" , "custom_view_artifacts" , "custom_views" } { |
| 497 | column := "view" |
| 498 | if table == "custom_views" { |
| 499 | column = "name" |
| 500 | } |
| 501 | if _, err := tx.ExecContext(ctx, "DELETE FROM " +table+" WHERE " +column+"=?" , view.Name); err != nil { |
| 502 | return err |
| 503 | } |
| 504 | } |
| 505 | return nil |
| 506 | }); err != nil { |
| 507 | return err |
| 508 | } |
| 509 | return t.deleteStoredArtifacts(ctx, ids) |
| 510 | } |
| 511 | |
| 512 | // customViewNameParam reads the name from a bare string or from {"name": ...}. |
| 513 | func customViewNameParam(params []json.RawMessage) string { |
| 514 | if len(params) == 0 { |
| 515 | return "" |
| 516 | } |
| 517 | if name := stringParam(params, 0); name != "" { |
| 518 | return name |
| 519 | } |
| 520 | var options struct { |
| 521 | Name string `json:"name"` |
| 522 | } |
| 523 | _ = json.Unmarshal(params[0], &options) |
| 524 | return strings.TrimSpace(options.Name) |
| 525 | } |
| 526 | |
| 527 | func customViewRefreshParam(params []json.RawMessage) bool { |
| 528 | if len(params) == 0 { |
| 529 | return false |
| 530 | } |
| 531 | var options struct { |
| 532 | Refresh bool `json:"refresh"` |
| 533 | } |
| 534 | if err := json.Unmarshal(params[0], &options); err != nil { |
| 535 | return false |
| 536 | } |
| 537 | return options.Refresh |
| 538 | } |
| 539 | |
| 540 | func (t *customViewService) customViewByName(ctx context.Context, name string) (customView, error) { |
| 541 | rows, err := t.customViewRows(ctx, name) |
| 542 | if err != nil { |
| 543 | return customView{}, err |
| 544 | } |
| 545 | if len(rows) == 0 { |
| 546 | return customView{}, errors.New("not found: view" ) |
| 547 | } |
| 548 | return rows[0], nil |
| 549 | } |
| 550 | |
| 551 | // customViewRows lists definitions, narrowed to one name. |
| 552 | func (t *customViewService) customViewRows(ctx context.Context, names ...string) ([]customView, error) { |
| 553 | query := `SELECT name,kinds,transform,trigger,audience,languages,max_bytes,secret,generation,enabled,created_at,last_run_at,last_status,failures FROM custom_views` |
| 554 | args := []any{} |
| 555 | if len(names) > 0 { |
| 556 | query += ` WHERE name=?` |
| 557 | args = append(args, names[0]) |
| 558 | } |
| 559 | query += ` ORDER BY created_at,rowid` |
| 560 | rows, err := t.store.DB().QueryContext(ctx, query, args...) |
| 561 | if err != nil { |
| 562 | return nil, fmt.Errorf("list custom views: %w" , err) |
| 563 | } |
| 564 | defer rows.Close() |
| 565 | var out []customView |
| 566 | for rows.Next() { |
| 567 | var item customView |
| 568 | var kinds, languages string |
| 569 | var enabled int |
| 570 | if err := rows.Scan(&item.Name, &kinds, &item.Transform, &item.Trigger, &item.Audience, &languages, &item.MaxBytes, &item.secret, &item.generation, &enabled, &item.CreatedAt, &item.LastRunAt, &item.LastStatus, &item.Failures); err != nil { |
| 571 | return nil, err |
| 572 | } |
| 573 | item.Enabled = enabled != 0 |
| 574 | if err := json.Unmarshal([]byte(kinds), &item.Kinds); err != nil { |
| 575 | return nil, fmt.Errorf("custom view kinds: %w" , err) |
| 576 | } |
| 577 | if err := json.Unmarshal([]byte(languages), &item.Languages); err != nil { |
| 578 | return nil, fmt.Errorf("custom view languages: %w" , err) |
| 579 | } |
| 580 | out = append(out, item) |
| 581 | } |
| 582 | return out, rows.Err() |
| 583 | } |
| 584 | |
| 585 | // customViewIndex returns the enabled views indexed by kind, built on first |
| 586 | // use and dropped whenever a view changes. |
| 587 | func (t *customViewService) customViewIndex(ctx context.Context) map[int][]customView { |
| 588 | t.viewMu.Lock() |
| 589 | defer t.viewMu.Unlock() |
| 590 | if t.viewIndex != nil { |
| 591 | return t.viewIndex |
| 592 | } |
| 593 | index := map[int][]customView{} |
| 594 | rows, err := t.customViewRows(ctx) |
| 595 | if err != nil { |
| 596 | t.telemetry.Logger().Error("load custom views" , "tenant" , t.tenantName, "error" , err) |
| 597 | return index |
| 598 | } |
| 599 | for _, row := range rows { |
| 600 | if !row.Enabled { |
| 601 | continue |
| 602 | } |
| 603 | for _, kind := range row.Kinds { |
| 604 | index[kind] = append(index[kind], row) |
| 605 | } |
| 606 | } |
| 607 | t.viewIndex = index |
| 608 | return index |
| 609 | } |
| 610 | |
| 611 | func (t *customViewService) invalidateCustomViews() { |
| 612 | t.viewMu.Lock() |
| 613 | t.viewIndex = nil |
| 614 | t.viewMu.Unlock() |
| 615 | } |
| 616 | |
| 617 | // CustomViews is the summary the web UI renders with: the enabled views' |
| 618 | // names and languages, nothing else. |
| 619 | func (t *customViewService) CustomViews() []views.View { |
| 620 | seen := map[string]bool{} |
| 621 | var out []views.View |
| 622 | for _, list := range t.customViewIndex(context.Background()) { |
| 623 | for _, view := range list { |
| 624 | if !seen[view.Name] { |
| 625 | seen[view.Name] = true |
| 626 | out = append(out, views.View{Name: view.Name, Languages: append([]string(nil), view.Languages...)}) |
| 627 | } |
| 628 | } |
| 629 | } |
| 630 | sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) |
| 631 | return out |
| 632 | } |
| 633 | |
| 634 | // queueCustomViews runs after an event is stored or released. Each enabled |
| 635 | // write-triggered view that watches the kind and finds a block it renders |
| 636 | // gets one transform intent. It never fails the publish. |
| 637 | func (t *customViewService) queueCustomViews(ctx context.Context, e event.Event) { |
| 638 | intents := t.PrepareReleased(ctx, e) |
| 639 | if len(intents) == 0 { |
| 640 | return |
| 641 | } |
| 642 | err := t.store.WithTx(ctx, func(tx *sql.Tx) error { |
| 643 | return storage.AddIntents(ctx, tx, intents, time.Now().Unix()) |
| 644 | }) |
| 645 | if err != nil { |
| 646 | t.telemetry.Logger().Error("view transforms not queued" , "tenant" , t.tenantName, "error" , err) |
| 647 | return |
| 648 | } |
| 649 | t.telemetry.Logger().Debug("view transforms queued" , "tenant" , t.tenantName, "queued" , len(intents)) |
| 650 | } |
| 651 | |
| 652 | // Prepare returns write-triggered transform intents for an event. The caller |
| 653 | // owns the transaction that persists the returned intents, allowing event and |
| 654 | // follow-up work to share one commit boundary. |
| 655 | func (t *customViewService) Prepare(ctx context.Context, e event.Event) []storage.Intent { |
| 656 | targets, err := t.CandidateTargets(ctx, e) |
| 657 | if err != nil { |
| 658 | t.telemetry.Logger().Error("plan custom view transforms" , "tenant" , t.tenantName, "error" , err) |
| 659 | return nil |
| 660 | } |
| 661 | intents, err := t.PrepareTargets(ctx, e, targets, false) |
| 662 | if err != nil { |
| 663 | t.telemetry.Logger().Error("prepare custom view transforms" , "tenant" , t.tenantName, "error" , err) |
| 664 | return nil |
| 665 | } |
| 666 | return intents |
| 667 | } |
| 668 | |
| 669 | // PrepareReleased returns intents after repository promotion has made Git |
| 670 | // objects available. Repository state events are intentionally accepted only |
| 671 | // through this path. |
| 672 | func (t *customViewService) PrepareReleased(ctx context.Context, e event.Event) []storage.Intent { |
| 673 | targets, err := t.CandidateTargets(ctx, e) |
| 674 | if err != nil { |
| 675 | t.telemetry.Logger().Error("plan released custom view transforms" , "tenant" , t.tenantName, "error" , err) |
| 676 | return nil |
| 677 | } |
| 678 | intents, err := t.PrepareTargets(ctx, e, targets, true) |
| 679 | if err != nil { |
| 680 | t.telemetry.Logger().Error("prepare released custom view transforms" , "tenant" , t.tenantName, "error" , err) |
| 681 | return nil |
| 682 | } |
| 683 | return intents |
| 684 | } |
| 685 | |
| 686 | // CandidateTargets captures names and registration identities for an event. |
| 687 | // The identity prevents a later registration with the same name from |
| 688 | // receiving work planned for an earlier registration. |
| 689 | func (t *customViewService) CandidateTargets(ctx context.Context, e event.Event) ([]customViewTarget, error) { |
| 690 | if eventExpired(e, time.Now().Unix()) { |
| 691 | return nil, nil |
| 692 | } |
| 693 | rows, err := t.customViewRows(ctx) |
| 694 | if err != nil { |
| 695 | return nil, err |
| 696 | } |
| 697 | targets := make([]customViewTarget, 0, len(rows)) |
| 698 | for _, view := range rows { |
| 699 | if !view.Enabled || view.Trigger != "write" || !view.watches(e.Kind) { |
| 700 | continue |
| 701 | } |
| 702 | if e.Kind != event.KIND_REPO_STATE && len(views.Matching(views.Blocks(e.Content), view.Languages)) == 0 { |
| 703 | continue |
| 704 | } |
| 705 | targets = append(targets, customViewTarget{Name: view.Name, Fingerprint: customViewFingerprint(view)}) |
| 706 | } |
| 707 | return targets, nil |
| 708 | } |
| 709 | |
| 710 | // PrepareTargets rechecks captured registration identities before creating |
| 711 | // intents. A removed and recreated view with the same name is rejected. |
| 712 | func (t *customViewService) PrepareTargets(ctx context.Context, e event.Event, targets []customViewTarget, released bool) ([]storage.Intent, error) { |
| 713 | if len(targets) == 0 || eventExpired(e, time.Now().Unix()) || (!released && e.Kind == event.KIND_REPO_STATE) { |
| 714 | return nil, nil |
| 715 | } |
| 716 | if e.Kind != event.KIND_REPO_STATE { |
| 717 | current, err := t.sourceCurrent(ctx, e) |
| 718 | if err != nil { |
| 719 | return nil, err |
| 720 | } |
| 721 | if !current { |
| 722 | return nil, nil |
| 723 | } |
| 724 | } |
| 725 | rows, err := t.customViewRows(ctx) |
| 726 | if err != nil { |
| 727 | return nil, err |
| 728 | } |
| 729 | candidates := make(map[string]customView, len(rows)) |
| 730 | for _, view := range rows { |
| 731 | candidates[view.Name] = view |
| 732 | } |
| 733 | if e.Kind == event.KIND_REPO_STATE { |
| 734 | if t.git == nil || t.statePending(ctx, e.ID) { |
| 735 | return nil, nil |
| 736 | } |
| 737 | } |
| 738 | var intents []storage.Intent |
| 739 | for _, target := range targets { |
| 740 | view, ok := candidates[target.Name] |
| 741 | if !ok || !view.Enabled || view.Trigger != "write" || !view.watches(e.Kind) { |
| 742 | continue |
| 743 | } |
| 744 | if customViewFingerprint(view) != target.Fingerprint { |
| 745 | continue |
| 746 | } |
| 747 | if e.Kind != event.KIND_REPO_STATE && len(views.Matching(views.Blocks(e.Content), view.Languages)) == 0 { |
| 748 | continue |
| 749 | } |
| 750 | intents = append(intents, t.viewIntent(view, e, "" , false)) |
| 751 | } |
| 752 | return intents, nil |
| 753 | } |
| 754 | |
| 755 | func eventExpired(e event.Event, now int64) bool { |
| 756 | expires := event.Expiration(e) |
| 757 | return expires > 0 && expires <= now |
| 758 | } |
| 759 | |
| 760 | // viewIntent is the transform intent for one source. The event travels in |
| 761 | // the payload; run makes a backfill or an hourly pass distinct from the |
| 762 | // write-time intent for the same event. |
| 763 | func (t *customViewService) viewIntent(view customView, e event.Event, run string, refresh ...bool) storage.Intent { |
| 764 | force := len(refresh) > 0 && refresh[0] |
| 765 | payload, _ := json.Marshal(viewPayload{Event: e, Attempt: 1, Run: run, Refresh: force, Generation: view.generation}) |
| 766 | id := e.ID |
| 767 | if run != "" { |
| 768 | id += "@" + run |
| 769 | } |
| 770 | return storage.Intent{Kind: viewTransform, EventID: id, Target: view.Name, Payload: string(payload)} |
| 771 | } |
| 772 | |
| 773 | // statePending reports whether a repository state still waits for its |
| 774 | // objects. |
| 775 | func (t *customViewService) statePending(ctx context.Context, id string) bool { |
| 776 | var one int |
| 777 | err := t.store.DB().QueryRowContext(ctx, `SELECT 1 FROM pending_events WHERE id=?` , id).Scan(&one) |
| 778 | return err == nil || t.git.IsPending(id) |
| 779 | } |
| 780 | |
| 781 | // queueCustomViewBackfill queues the newest sources of a view's kinds, at |
| 782 | // most 500, written since the given time. Repository states are queued |
| 783 | // whether or not their README has blocks; the handler reads it. |
| 784 | func (t *customViewService) queueCustomViewBackfill(ctx context.Context, view customView, since int64, run string, refresh ...bool) (int, error) { |
| 785 | force := len(refresh) > 0 && refresh[0] |
| 786 | rows, err := t.store.Query(ctx, event.Filter{Kinds: view.Kinds, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: viewBackfillLimit}) |
| 787 | if err != nil { |
| 788 | return 0, err |
| 789 | } |
| 790 | var intents []storage.Intent |
| 791 | for _, e := range rows.Events { |
| 792 | if e.CreatedAt < since { |
| 793 | continue |
| 794 | } |
| 795 | if e.Kind == event.KIND_REPO_STATE { |
| 796 | if t.git == nil || t.statePending(ctx, e.ID) { |
| 797 | continue |
| 798 | } |
| 799 | } else if len(views.Matching(views.Blocks(e.Content), view.Languages)) == 0 { |
| 800 | continue |
| 801 | } |
| 802 | intents = append(intents, t.viewIntent(view, e, run, force)) |
| 803 | } |
| 804 | if len(intents) == 0 { |
| 805 | return 0, nil |
| 806 | } |
| 807 | err = t.store.WithTx(ctx, func(tx *sql.Tx) error { |
| 808 | return storage.AddIntents(ctx, tx, intents, time.Now().Unix()) |
| 809 | }) |
| 810 | return len(intents), err |
| 811 | } |
| 812 | |
| 813 | // tickCustomViews queues the hourly views that are due. It is called from |
| 814 | // the tenant scheduler on its own timer. |
| 815 | func (t *customViewService) tickCustomViews(ctx context.Context, now int64) error { |
| 816 | rows, err := t.customViewRows(ctx) |
| 817 | if err != nil { |
| 818 | return err |
| 819 | } |
| 820 | for _, view := range rows { |
| 821 | if !view.Enabled || view.Trigger != "hourly" || now-view.LastRunAt < viewHourlyPeriod { |
| 822 | continue |
| 823 | } |
| 824 | since := view.LastRunAt - 300 |
| 825 | if view.LastRunAt == 0 { |
| 826 | since = 0 |
| 827 | } |
| 828 | queued, err := t.queueCustomViewBackfill(ctx, view, since, strconv.FormatInt(now/viewHourlyPeriod, 10), false) |
| 829 | if err != nil { |
| 830 | return err |
| 831 | } |
| 832 | if _, err := t.store.DB().ExecContext(ctx, `UPDATE custom_views SET last_run_at=? WHERE name=?` , now, view.Name); err != nil { |
| 833 | return err |
| 834 | } |
| 835 | if queued > 0 { |
| 836 | t.telemetry.Logger().Info("custom view hourly run queued" , "tenant" , t.tenantName, "queued" , queued) |
| 837 | } |
| 838 | } |
| 839 | return nil |
| 840 | } |