contentkit

contentkit

contentkit is the deterministic content library for the Doujins, Hentai0 and marketplace hosts: tenant-scoped interactions (posts, comments, reactions, favorites, polls), keyword search over host content, the ClickHouse signal plane (consumption, feedback, exposures, popularity, erasure) and the discovery reads over both. It needs no model provider, API key or vector extension. Semantic search belongs entirely to the separate, deferred User Intelligence library; ContentKit has no semantic search configuration or runtime hook.

Design: open-rails-tracker/contentkit/DESIGN.md. Host contract: HOST_INTEGRATION.md. Migrations: docs/migration.md, docs/restore.md.

Release policy

ContentKit is pre-stable and releases on v0.x; its API may change between minor versions. The historical v1.0.0-rc.1 through v1.1.1 tags were premature and are retracted. v1.1.2 is a self-retracted, withdrawal-only marker carrying Go module metadata, not a supported stable API release.

The marker and v0.12.2 identify the same source commit. Go reads retractions from the highest release before applying them, so the marker makes fresh @latest requests select the latest unretracted v0.x release, including future v0.x releases. Existing tags remain immutable; retraction preserves explicit pins and does not automatically downgrade existing consumers. See Go module retractions.

Vocabulary

A language is metadata on a document or version, never part of a reference. Every read is scoped to the tenant pinned at construction; a reference of another tenant is an error, never remapped.

Packages

Install

One call installs all PostgreSQL features in a host-selected schema (which may also hold application tables) and the signal plane in ClickHouse. PGroonga and pg_trgm live in public; no vector extension is required:

_ = signal.CreateDatabase(ctx, adminCH, "hub", cluster)
_ = contentkit.Migrate(ctx, contentkit.MigrateConfig{
	DB: sqlDB, Schema: "doujins",
	ClickHouse: &chmigrate.Config{ClientAddr: addr, Database: "hub", App: "contentkit_signal", Cluster: cluster},
})

The baselines initialize fresh stores; newer PostgreSQL migrations upgrade the current lineage, not retired migration chains. See docs/migration.md.

Runtime

rt, _ := contentkit.NewRuntime(ctx, contentkit.RuntimeConfig{
	EmbeddedConfig: contentkit.EmbeddedConfig{PG: pool, PGSchema: "doujins", Tenant: "doujins", CH: ch, CHDatabase: "hub"},
	Content: content.Options{Schema: "doujins", Identity: identity, Authz: authz, Resolver: resolver, ContentKinds: []string{"gallery", "post"}},
})
mux.Handle("/api/social/", http.StripPrefix("/api/social", rt.Handler()))
counts, _ := rt.Content.Counts(ctx, []contentkit.ContentRef{rt.Content.Ref("gallery", "42")})
_ = worker.SyncOnce(ctx, rt.WorkerOptions(hostWorkerOptions)) // host documents + posts

rt is the Hub (search, typeahead, signals, discovery) plus rt.Content (interactions). See HOST_INTEGRATION.md.

Search

Documents are one per content reference and language: a gallery with an English original, an English colored edition and a Spanish original is three documents keyed by version. Hosts mark changes in the transaction that changes content and run one worker tick per schedule:

_ = search.MarkDirty(ctx, tx, schema, []search.DirtyMark{{DocumentKey: search.DocumentKey{ContentRef: ref, Language: "en"}}})

_ = worker.SyncOnce(ctx, worker.Options{
	Pool: pool, Schema: schema, Tenant: "doujins",
	SupportedLanguages: []string{"en", "es"}, ContentKinds: []string{"gallery"},
	ListContent:           listGalleries,          // bounded pages of refs, for backfill
	BuildKeywordDocuments: buildGalleryDocuments,  // refs -> KeywordDocument{Title, Aliases, Keywords}
	Sink:                  nil,                    // optional DocumentSink
})

The worker records rejected host documents in content_search_invalid with their dirty-queue revision, validation error and failure time. It continues with other documents and backfill, but does not acknowledge the rejected row. Inspect that table for repair; marking the document dirty after correcting its source advances the revision and retries it. A deletion clears the invalid record. Run the PostgreSQL migrations before starting a worker built against this schema.

Querying groups documents per work before paging; the host's eligibility join decides, per document, whether that one row is visible and which is preferred:

client, _ := contentkit.NewClient(contentkit.ClientConfig{Pool: pool, Schema: schema, Tenant: "doujins"})
page, _ := client.Search(ctx, query, contentkit.SearchOptions{
	Language: "es", ContentKinds: []string{"gallery"}, Limit: 20,
	Eligibility: &contentkit.Eligibility{SQL: eligibilitySQL, Args: args},
})
for _, hit := range page.Hits { /* hit.ContentID (work), hit.Version(), hit.Language, hit.Score */ }

SearchHit.Score is the keyword match tier (exact title 1, alias 0.9, token/prefix 0.75–0.77, typo 0.5–0.52). SearchWithTrace returns retrieval provenance for evaluation. Query limits: 256 characters, 16 tokens, a two-second ceiling per request.

Ports

DocumentSink is a neutral document change feed for external indexes, caches or audit consumers. ContentKit ships no sink implementation or AI-specific configuration.

Errors

Both mounted handlers (content.Runtime.Handler, taxonomy.Handler) answer failures with one flat body. Branch on code; error is a human message and may change.

{"error":"not found","code":"not_found"}

5xx bodies carry no cause: it goes to Options.Logger (slog.Default() when unset) with the request method, path, status and duration. Postgres constraint names, driver text and stack traces are logged, never served.

Media

Media is an app-defined, self-describing file system in one private bucket; no database table records what media exists. The app declares its kinds in one registry (media.Config); HOST_INTEGRATION "Media" covers the registry, commit ops, reads and exposure.

{namespace}/{kind}/{id}/manifest.json         gzip JSON: the ordered file list with provenance; never served
                       /private/sha256-{hex}  every blob: uploads, derived files, editor views (token)
                       /public/{name}         app-declared names, e.g. cover-460.webp (anyone)
                       /temp/{name}           staged uploads (u-{uuid}) and in-flight writes; never served
{namespace}/{kind}/_default/public/{name}     a public preset's default image

An item is the host's version (gallery/456 English, gallery/789 Korean). Private blobs are content-addressed and immutable; public names are fixed and overwritten in place (ETags and a CDN purge keep them fresh). Every URL is https://media.<site>/v1/{namespace}/{kind}/{id}/{public|private}/{name}.

reg, _ := media.NewRegistry(media.Config{Namespace: "doujins", BaseURL: "https://media.doujins.ai",
	Kinds: []media.Kind{Gallery, accountmedia.User},
	Hooks: media.Hooks{Resolver: resolver, CanUpload: authorizer, PurgePublic: purge, ItemReady: ready}})
store, _ := s3.New(s3.Config{Bucket: "media", Endpoint: rgw, PublicEndpoint: "https://s3.doujins.ai", UsePathStyle: true,
	AccessKeyID: id, SecretAccessKey: secret})
key, _ := token.ParseKey(os.Getenv("MEDIA_TOKEN_KEY")) // "{kid}:{base64}", shared with media-access
_ = workqueue.Migrate(ctx, pool, "doujins_media_worker") // this host's worker schema
queue, _ := workqueue.New(pool, reg, "doujins_media_worker")
jobs, _ := media.NewJobs(media.JobsConfig{Store: store, Registry: reg, Locker: media.PGLocker(pool), Pool: pool,
	Processes: queue, Limiter: limiter})
uploads, _ := media.NewUploads(media.UploadOptions{Store: store, Manifests: jobs.Manifests(), Tickets: &ring,
	Limiter: limiter, Queue: queue, Frames: frames})
reader, _ := media.NewReader(media.ReaderOptions{Manifests: jobs.Manifests(), Queue: queue, Progress: progress,
	Delivery: media.Delivery{Mode: media.DeliverCookie, CookieDomain: "doujins.ai", SigningKey: key}})
mux.Handle("/api/media/upload/", http.StripPrefix("/api/media/upload", media.UploadHandler(uploads, media.UploadHandlerOptions{Actor: actorOf})))
mux.Handle("/api/media/", http.StripPrefix("/api/media", reader.Handler(media.HandlerOptions{Identity: identity})))
// composed into the host's River client: jobs.RiverJobs()
_ = jobs.ExposeTx(ctx, tx, ref)                                          // whenever anonymous visibility changes
_ = jobs.DeleteItemsTx(ctx, tx, media.Deletion{Ref: ref, Owner: owner}) // in the host's delete transaction

Manifests. Every edit runs under the Locker (required: PGLocker, a Postgres advisory lock shared by every process on the bucket) and writes with If-Match (or If-None-Match: *) once the store reports conditional PUT, retrying on conflict. The edit is normalized (canonical order, download names, dropped originals) and validated; an unchanged manifest is not written. Reads go through an in-process cache bounded by bytes and revalidated by ETag, so a read is never stale. A 2,000-page gallery is about 1.5 MB of JSON and 400 KB stored; a 2-hour video about 5 KB, its segment tables living in index blobs. Every read and write stops at MaxManifestBytes (8 MiB of JSON; the default 128 MiB cache holds five). Edits stop 4 KiB short of it and only when they grow the manifest, so a full item can still shrink and the hidden and full flags always fit. A commit is refused (413 too_large) when the item, once processed, would pass that: it projects every output its presets have yet to write. When outputs still overrun it, the worker marks the item full instead of recording them: producers then write no private output for it (its public files still render), and editor reads report full (state full), until a commit frees the bytes the refused record was short of (deficit, counting what removed uploads would still have added). An item holds at most MaxUploads (10,000) uploads, an upload's meta at most 4 KiB of JSON and the item's 16 KiB; names are capped at 200 bytes of JSON, written unescaped (< is one byte). The manifest object records the quota its uploads were charged, so deleting an item never needs it to decode, and hiding deletes public/ before touching it.

The bucket is optional at startup. s3.New never dials; register store.Check(ctx, prefix) as the host's optional S3 dependency probe. Its first success probes the backend's capabilities unless Config.Capabilities declares them; a throttled or cut-off probe records nothing. An unreachable or 5xx bucket is media.ErrUnavailable: 503 unavailable over HTTP, and in media jobs a River snooze (not an attempt) while a fresh Check confirms the outage, capped by MaxOutageSnoozes (media.SnoozeUnavailable).

Uploads go straight to the bucket, to a staged name: up to 64 MiB is one PUT to temp/u-{uuid} signed with its type, length and SHA-256, larger files are multipart (8–16 MiB parts, each signed with its length and SHA-256, resumed through ListParts, completed by the server from a signed ticket; nothing is stored). The browser's hash only finds an identical blob already in the folder (no upload). After the commit, the worker's place job hashes each staged upload and writes it to private/sha256-{hex} of the bytes it read (Manifests.Place); an existing blob is never overwritten, so a blob always holds the bytes of its name. The optional UploadLimiter (media.NewPGLimiter) rate-limits uploaders and charges each upload's size (at least 64 KiB) to the grant's Owner; growth past the quota fails with 413 quota_exceeded, and deleting an item releases it.

Access agent (cmd/media-access, media/agent, image ghcr.io/open-rails/contentkit-media-access:{tag}): public/ to anyone (public, max-age=300, stale-while-revalidate=86400, falling back to the kind's _default for declared names), private/sha256-{hex} with the item's token in ?t= or the mt cookie (private, immutable). A token opens every private file of its item or none; ?dl={name} serves the file as a download under that name when it has the file's type. Everything else and every denial is one no-store 404. It keeps no state: rate limit the media host at the ingress; the read API limits the items a viewer opens per hour (ReaderOptions.Issuance). It needs MEDIA_ACCESS_S3_ENDPOINT, _S3_BUCKET, a key reading only */private/* and */public/*, MEDIA_ACCESS_TOKEN_KEY and _TOKEN_KEY_PREVIOUS, MEDIA_ACCESS_HOSTS (media.doujins.ai=doujins,accounts; media.hanime.media=hentai0,accounts), MEDIA_ACCESS_CORS_ORIGINS and MEDIA_ACCESS_DEFAULTS (layout.FormatDefaults(media.AgentConfig(reg).Defaults)).

The media worker (media/worker) runs every producer from the host's worker River schema (media/workqueue, MEDIA_WORKER_SCHEMA, per host: hosts sharing a database never share one): images, zips, public presets and editor views (media/image, libvips) and HLS, MP4, audio, subtitles and frames (media/video, ffmpeg). The host presigns, commits, exposes and reads, and links only media/workqueue. The worker hands readiness (Hooks.ItemReady), purges (Hooks.PurgePublic) and sweeps back to the host's River schema (MEDIA_HOST_RIVER_SCHEMA), so the stock build (cmd/media-worker, image ghcr.io/open-rails/contentkit-media-worker) needs only the registry as JSON (MEDIA_KINDS_FILE). Hosts with a Private.Choose build their own:

cfg, _ := worker.FromEnv(ctx) // DATABASE_URL, MEDIA_S3_*, MEDIA_WORKER_SCHEMA, MEDIA_HOST_RIVER_SCHEMA, MEDIA_WORKER_*
cfg.Kinds = reg                // the host's registry
w, _ := worker.New(ctx, cfg)   // no DDL: the host's migrate step runs workqueue.Migrate
_ = w.Run(ctx)                 // until SIGTERM; running jobs get MEDIA_WORKER_SHUTDOWN_GRACE

It never exits for a missing dependency: Run takes no jobs until the bucket answers, and MEDIA_METRICS_ADDR serves /livez, /readyz, /statusz and app_dependency_up with /metrics.

Images (media/image, CGO over libvips): WebP at each Image spec (inside or cover box, quality, blur), through the upload's edit (crop in EXIF-oriented source pixels, then a clockwise quarter rotation); nothing is upscaled. GIF and WebP animations keep every frame, delay and loop count; MaxPixels (100 MP over all frames), MaxFrames (1000) and MaxAnimationSeconds (60) bound sources, and a stored source over its Upload's MaxBytes is refused unread. Uploads feeding image presets take only media.ImageTypes (JPEG, PNG, WebP, GIF, AVIF, HEIF, TIFF), the declared type binds the decoder, and every other libvips loader is blocked (libvips 8.13 or later). Public presets render every width through the same spec (a width past the edited image at its width) and carry from and fp as object metadata.

Video (media/video): an HLS preset encodes the ladder (rung = short side, default 2160/1080/480, none above the source) in each of the worker's codecs (MEDIA_WORKER_CODECS, default av1,h264), plus AAC per audio track, WebVTT per text subtitle and a seek sprite. Each rendition is one byte-range fMP4 blob whose segment table is its own index blob; stages are published rung by rung (the upload stays pending until the last), a compliant top rung is stream-copied, and sources outside the aspect bounds fail. An MP4 preset muxes H.264 at one rung with the default audio; Audio gives an HLS track and an M4A (optional EBU R128 loudness); Subtitles converts SRT and SSA/ASS to clean WebVTT. Playlists are built per request by the read API. Encode progress comes from workqueue.NewProgressSource (ReaderOptions.Progress).

Jobs (media.Jobs, composed into the host's River client): the sweep collects garbage by manifest reference (unreferenced private blobs a grace period old, unexpected public names at once, temp/ by age); folder deletion (with a late-upload second pass and quota release); Expose; Regenerate; SweepOrphans; and the worker's relays.

Tiered access (media/tiered) maps public/members/ppv/ members_ppv/premium to Resolutions over an entitlement Checker.

Browser SDK (sdk/upload, @openrails/contentkit-upload, attached to each release): hashing, uploads, commit ops, reads, waitFor, the frame picker, publicURL/srcSet, React hooks and UI. See sdk/upload/README.md.

Taxonomy

Nodes, names, edges and assignments are tenant-scoped; effective tags are the work's assignments ∪ the selected version's; RequireAll makes a multi-node filter hold on one eligible version inside the same join as search. The PostgreSQL baseline always installs the taxonomy tables; see docs/taxonomy-migration.md.

store, _ := taxonomy.New(taxonomy.Options{Pool: pool, Schema: schema, Tenant: "doujins",
	Kinds: []string{"tag", "artist", "character", "series", "voice_actor"}, Languages: []string{"en", "es"},
	CountEligibility: &search.Eligibility{SQL: releasedVersionSQL}})
_ = store.WithTx(tx).Assign(ctx, []taxonomy.Assignment{{ContentRef: g1.WithVersion(v2), TaxonomyID: "colored"}}, taxonomy.AssignOptions{})
tags, _ := store.EffectiveTags(ctx, []contentkit.ContentRef{g1.WithVersion(v2)})
filter, args, _ := taxonomy.RequireAll(schema, []taxonomy.TaxonomyID{"colored"})
page, _ := client.Search(ctx, q, contentkit.SearchOptions{Language: "es", ContentKinds: []string{"gallery"}, FilterSQL: filter, FilterArgs: args, Eligibility: elig})
mux.Handle("/admin/taxonomy/", http.StripPrefix("/admin/taxonomy", taxonomy.Handler(store)))

ListNodes backs a catalog index page directly: the display name in the request language (falling back to the store's configured order, or pinned with LanguageMode), the per-language content count, an A-Z index, a name+alias search, hide-empty, five orders and offset paging with a total. The zero ListOptions keeps the keyset contract a full admin sync wants.

page, _ := store.ListNodes(ctx, taxonomy.ListOptions{Kind: "artist", Language: "es",
	ContentKind: "gallery", MinCount: 1, Sort: taxonomy.SortCount, Offset: 40, Limit: 20})
// page.Total, and per row: Name, NameLanguage, Count.
letter, _ := store.ListNodes(ctx, taxonomy.ListOptions{Kind: "artist", Language: "es", NamePrefix: "a"})
cast, _ := store.ListNodes(ctx, taxonomy.ListOptions{Kind: "character", Related: "s-fate", Relation: taxonomy.RelationMemberOf})

Worker: ContentKinds: append(hostKinds, store.Kinds()...), ListContent: store.Lister(listGalleries), BuildKeywordDocuments: store.Builder(buildGalleryDocuments).

Signal plane and discovery

hub, _ := contentkit.NewEmbedded(contentkit.EmbeddedConfig{
	PG: pool, PGSchema: schema, CH: ch, CHDatabase: "hub", Tenant: "doujins",
	Scorers:  map[string]signal.Scorer{"gallery": galleryScorer},
	Catalogs: map[string]contentkit.ContentCatalog{"gallery": galleryCatalog},
})
g1 := hub.Content("gallery", "g1")
_ = hub.RecordSignals(ctx, []signal.Signal{{ContentRef: g1, Subject: user, Type: signal.TypeView, EventID: sessionID, Revision: checkpoint, OccurredAt: start, Progress: 95, ProgressMax: 100}})
states, _ := hub.States(ctx, user, []contentkit.ContentRef{g1, g1.WithVersion("v2")})
top, _ := hub.Popular(ctx, "gallery", signal.PopularOptions{Window: signal.LastDays(30, time.Now())})

• Identity is (tenant, content ref, subject, type, event id); retries, reordering and revisions converge, nothing is incremented. Record the work view and, separately, the selected version's view (g1.WithVersion(v)).
• Work reads (History, Popular, SeenIDs, co-engagement) count the work once across versions; States and Metrics read exactly the references given, work or version.
• Windows are whole UTC days, 7/30/90/365/all, no decay.
• Rank by a named policy, not the default rank: popularity.ByName("v1"), popularity.New(popularity.Config{Source: hub, Policy: policy}), then ranker.Popular / ranker.Scores / ranker.Taxonomy (docs/popularity-policy.md).
• SimilarTo and Recommend draw candidates from EmbeddedConfig.Candidates (default: co-engagement); see discovery candidates.
• EraseSubjects is account erasure with a quorum-written fence; see HOST_INTEGRATION.md and, for interaction data, interaction erasure.
• Reactions and favorites reach the signal plane from their own revisioned rows: schedule rt.SyncPreferences (watermark with a commit overlap) and rt.ResyncPreferences (full re-send); see HOST_INTEGRATION.md.

Testing

CONTENTKIT_TEST_URL=postgres://...  CONTENTKIT_PROFILE_URL=postgres://... \
CONTENTKIT_TEST_CH_ADDR=localhost:9000 CONTENTKIT_TEST_CH_USER=... CONTENTKIT_TEST_CH_PASSWORD=... CONTENTKIT_TEST_CH_CLUSTER=... \
go test ./... -race -count=1 -p 2

Media tests also need an S3 backend (CONTENTKIT_TEST_S3_ENDPOINT, _ACCESS_KEY, _SECRET_KEY, optional _REGION, _BUCKET and _REQUIRE; see media/internal/s3test). CI runs them on MinIO; media/image runs in its own CI job with libvips, and the other jobs exclude it. To record a Ceph RGW release's capabilities, point the same variables at an RGW bucket and run go test ./media/... -v -count=1; the log prints the probed capabilities. Manifest edits run under PGLocker, so the tests also need CONTENTKIT_TEST_URL (they skip without it). media/video tests also need ffmpeg and ffprobe on PATH (they skip without them unless CONTENTKIT_TEST_FFMPEG=1).

Tests run against real PGroonga Postgres and ClickHouse+Keeper and skip without the variables; CONTENTKIT_PROFILE_URL needs CREATEDB. Regenerate the eval baseline with CONTENTKIT_EVAL_UPDATE=1.