library-operator

1package main23// The wire types are hand-written, the way liken and the sibling4// operators write theirs. The Kubernetes API is HTTPS that serves5// JSON, and importing client-go for a dozen structs brings informers,6// work queues, and a release cadence this program does not use. Each7// type carries only the fields this operator reads or writes; the API8// server fills in the rest.910import (11	"slices"12	"time"13)1415// The group this operator serves, and the core group it writes16// into: a Library becomes an ordinary CronJob and its Jobs ordinary17// pods, so any tool that reads them reads what a Library became.18const (19	libraryAPIVersion = "library.liken.sh/v1alpha1"20	podAPIVersion     = "v1"21)2223// The group media-operator serves. This operator reads Players from it24// and writes none, so the group appears here for the read path alone.25const playerAPIVersion = "media.liken.sh/v1alpha1"2627// The finalizer this operator holds on every Library. It keeps a28// deleted Library open until the departure in depart.go has swept29// the library's rows out of every surviving agent's catalog. The30// name is in this operator's own group, so the finalizer says which31// controller answers for it.32const libraryFinalizer = "library.liken.sh/cleanup-library"3334// Release 2026.08.31-004 named the finalizer sweep. The operator35// still reads that name so a Library which adopted it under that36// release swaps to the current name on a pass, and deletes instead37// of sticking on a finalizer nothing releases.38const formerLibraryFinalizer = "library.liken.sh/sweep"3940// ObjectMeta carries what this operator reads or writes: name and41// namespace for the URL, resourceVersion for the conditional write,42// uid with ownerReferences so the garbage collector takes a scanner43// pod with its Library, and labels so one watch selects every scanner44// pod in the cluster.45//46// Annotations carry the template hash the operator stamps on a scanner47// pod, which is how a pass tells a live pod from the pod it would48// build now. deletionTimestamp is set by the API server on an object49// on its way out, and a pod with one set is left alone until the50// delete completes. finalizers is here because the operator holds a51// Library open past the delete request, and that window is where the52// departure in depart.go sweeps the catalog. generateName is the prefix53// the API server mints a name from; a Play carries it in place of a54// name, because every start of a title is its own Play and the operator55// keeps no record of what it created.56type ObjectMeta struct {57	Name              string            `json:"name,omitempty"`58	GenerateName      string            `json:"generateName,omitempty"`59	Namespace         string            `json:"namespace,omitempty"`60	UID               string            `json:"uid,omitempty"`61	Generation        int64             `json:"generation,omitempty"`62	ResourceVersion   string            `json:"resourceVersion,omitempty"`63	Labels            map[string]string `json:"labels,omitempty"`64	Annotations       map[string]string `json:"annotations,omitempty"`65	Finalizers        []string          `json:"finalizers,omitempty"`66	DeletionTimestamp string            `json:"deletionTimestamp,omitempty"`67	OwnerReferences   []OwnerReference  `json:"ownerReferences,omitempty"`68}6970// Deleting reports that somebody asked for this object's deletion.71// The API server does not remove an object that carries a finalizer;72// it writes the deletion timestamp instead, and that window is where73// the departure runs.74func (m ObjectMeta) deleting() bool { return m.DeletionTimestamp != "" }7576// Holds reports whether this object carries the named finalizer.77func (m ObjectMeta) holds(finalizer string) bool {78	return slices.Contains(m.Finalizers, finalizer)79}8081// With answers the finalizer list with one added, and without82// answers it with one removed. Both answer a new slice, because the83// caller's copy is the object as the server has it, and a patch that84// fails must leave that copy alone.85func (m ObjectMeta) with(finalizer string) []string {86	if m.holds(finalizer) {87		return m.Finalizers88	}89	return append(slices.Clone(m.Finalizers), finalizer)90}9192func (m ObjectMeta) without(finalizers ...string) []string {93	kept := []string{}94	for _, held := range m.Finalizers {95		if !slices.Contains(finalizers, held) {96			kept = append(kept, held)97		}98	}99	return kept100}101102// An ownerReference ties an object's life to its owner's: the garbage103// collector deletes the owned object when the owner goes, which is104// this operator's whole teardown. Controller is true because exactly105// one thing manages each owned object; there is no blockOwnerDeletion,106// because nothing here needs the owner to wait.107type OwnerReference struct {108	APIVersion string `json:"apiVersion"`109	Kind       string `json:"kind"`110	Name       string `json:"name"`111	UID        string `json:"uid"`112	Controller bool   `json:"controller"`113}114115// A list's own resourceVersion is the revision of the whole116// collection, which is what a watch resumes from.117type ListMeta struct {118	ResourceVersion string `json:"resourceVersion,omitempty"`119}120121// A Library is a volume of media of one kind. The operator reads the122// spec and writes the status: the spec is what a person declared, and123// the status is what the volume resolved to and what the scanner124// reports.125type Library struct {126	APIVersion string        `json:"apiVersion,omitempty"`127	Kind       string        `json:"kind,omitempty"`128	Metadata   ObjectMeta    `json:"metadata"`129	Spec       LibrarySpec   `json:"spec"`130	Status     LibraryStatus `json:"status"`131}132133type LibraryList struct {134	Metadata ListMeta  `json:"metadata"`135	Items    []Library `json:"items"`136}137138// A Library names its storage, its kind, and the settings for that139// kind. The settings blocks are pointers because their presence is the140// declaration: the block that matches the kind is there and no other,141// and the CRD's own rule refuses any other combination.142type LibrarySpec struct {143	Storage LibraryStorage   `json:"storage"`144	Kind    string           `json:"kind"`145	Movies  *LibrarySettings `json:"movies,omitempty"`146	Series  *LibrarySettings `json:"series,omitempty"`147	// Franchises is the settings block of the franchises kind: the image148	// every block carries, and the art claim that this kind alone names.149	Franchises *LibraryFranchises `json:"franchises,omitempty"`150151	// The metadata providers to ask about a title, in the order they152	// are asked. Enrichment reads the list; nothing acts on it yet.153	Sources []string `json:"sources,omitempty"`154155	// The path components the walk skips, wherever they sit under the156	// library root. An owner names the junk their storage keeps, such as157	// a recycle bin or a staging folder, because no fixed list can158	// anticipate every volume's layout.159	Ignore []string `json:"ignore,omitempty"`160161	// One time per fact, by the names status.gaps uses: an attempt162	// this fact made before that time does not count, so every title163	// is in that fact's gap again and the fact asks a provider again164	// and rewrites its own files and rows in place.165	// Nothing is deleted, and a fact this map does not name is166	// untouched.167	Refresh map[string]time.Time `json:"refresh,omitempty"`168169	// How often the full walk runs.170	Scan LibraryScan `json:"scan,omitzero"`171172	// Whether this library builds the thumbnail sheets a scrub bar reads, which173	// costs hours of CPU over a whole library on the first run.174	Trickplay LibraryTrickplay `json:"trickplay,omitzero"`175}176177// The trickplay block of the spec, off unless the owner turns it on.178type LibraryTrickplay struct {179	Enabled bool `json:"enabled,omitempty"`180}181182// The schedule the Library's full walk runs on, as the cron183// expression a CronJob takes.184type LibraryScan struct {185	Schedule string `json:"schedule,omitempty"`186}187188// Once an hour, which is the interval a library with a webhook189// needs as a backstop and a library with none can live on.190const defaultScanSchedule = "0 * * * *"191192// The schedule the CronJob takes: the Library's own, or the193// default when it names none. The CRD defaults the field, so a Library194// from the API server always carries one, and this answers for the ones195// built in this program.196func (s LibrarySpec) scanSchedule() string {197	if s.Scan.Schedule != "" {198		return s.Scan.Schedule199	}200	return defaultScanSchedule201}202203// The kinds of media a Library holds. Each one names a settings block204// on the spec and a scanner to run, and a new kind is a new block205// beside the ones here.206const (207	libraryKindMovies = "movies"208	libraryKindSeries = "series"209	// A franchises library walks the franchise.yaml files on its storage210	// claim, and writes the art it downloads into the second claim that211	// its settings block names.212	libraryKindFranchises = "franchises"213)214215// Settings is the block that matches the kind, which is the block the216// scanner receives. Keeping the choice here keeps the Go side and the217// CRD's rule saying the same thing, so a new kind is one more case218// here and one more clause there.219func (s LibrarySpec) settings() *LibrarySettings {220	switch s.Kind {221	case libraryKindMovies:222		return s.Movies223	case libraryKindSeries:224		return s.Series225	case libraryKindFranchises:226		if s.Franchises == nil {227			return nil228		}229		return &s.Franchises.LibrarySettings230	}231	return nil232}233234// artClaim is the claim a franchises scan writes its art into, and empty235// for every other kind. The checkout is read-only, so the art has to land236// on a claim of its own.237func (s LibrarySpec) artClaim() string {238	if s.Kind != libraryKindFranchises || s.Franchises == nil {239		return ""240	}241	return s.Franchises.Art.Claim242}243244// screenClaim is the claim that holds the files a screen reads: the art245// claim of a franchises library, and the storage claim of every other246// kind. The screen pod, the play request, and the enrich Job all ask it,247// so no one of them decides the question on its own.248func (s LibrarySpec) screenClaim() string {249	if claim := s.artClaim(); claim != "" {250		return claim251	}252	return s.Storage.Claim253}254255// screenRoot is the directory inside the screen's claim that holds this256// library's files. The storage root applies to the storage claim alone,257// and the art of a franchises library lands at the root of its own claim,258// so a screen and a play reference read the art from there.259func (s LibrarySpec) screenRoot() string {260	if s.artClaim() != "" {261		return "/"262	}263	return s.Storage.Root264}265266// LibraryStorage is the volume and the directory inside it. Every kind267// mounts the claim read-only, a franchises library included, because the268// claim holds the truth the scan walks and the scan writes nothing to it.269// The operator reads the PersistentVolume behind the claim for the270// volume's kind and address.271type LibraryStorage struct {272	Claim string `json:"claim,omitempty"`273274	// The directory inside the claim this library starts at, always an275	// absolute path from the root of the volume. The CRD defaults it276	// to / and refuses a relative path, so the scanner takes this277	// field as it stands.278	Root string `json:"root,omitempty"`279}280281// LibrarySettings is the one setting every kind's block carries. The282// movies and series blocks are this struct alone, and the franchises block283// embeds it beside the art claim.284type LibrarySettings struct {285	// The scanner image to run in place of the one the project ships286	// for the kind, which is how a person supplies a scanner of their287	// own. Empty means the operator's own image.288	Image string `json:"image,omitempty"`289}290291// LibraryFranchises is the settings block of the franchises kind: the292// image every block carries, and the art claim that this kind alone names.293// The CRD requires the art claim, because a franchises scan always294// downloads art and the checkout cannot take it.295type LibraryFranchises struct {296	LibrarySettings297	Art LibraryArt `json:"art,omitzero"`298}299300// LibraryArt names the writable claim, in the Library's namespace, that a301// franchises scan downloads the art into. It is the claim a screen mounts302// for this library, because the art is what a screen reads of a franchise.303type LibraryArt struct {304	Claim string `json:"claim,omitempty"`305}306307// LibraryStatus is what the operator reports on a Library: the volume308// the claim resolved to, the scanner's report, the pod that runs the309// scanner, and the conditions.310//311// The counts carry no omitempty. A library of zero titles is a312// real answer, and a column that reads 0 says it, where an omitted313// field reads as nothing at all. The conditions say whether a report314// arrived.315//316// RemovedLastSweep is the count of catalog rows the scanner's last full317// sweep removed, folded from the bus report. A partial walk that pruned318// too much shows here, so a mass delete is visible without a shell.319type LibraryStatus struct {320	Volume *LibraryVolume `json:"volume,omitempty"`321	// Phase says what the scanner is doing, in one word a person reads at322	// a glance: one of the four values below. Where Ready is the condition a323	// program matches on, Phase is the sentence a person reads.324	Phase        string `json:"phase,omitempty"`325	Titles       int    `json:"titles"`326	Unidentified int    `json:"unidentified"`327	// Items is how many item rows the catalog holds for this library,328	// across the movies, series, and episodes tables, and Files how many329	// file rows. Both are the catalog's own counts, read after the prune,330	// so they describe what a screen can read and not what one walk saw.331	Items            int       `json:"items"`332	Files            int       `json:"files"`333	RemovedLastSweep int       `json:"removedLastSweep"`334	LastWalk         time.Time `json:"lastWalk,omitzero"`335	LastChange       time.Time `json:"lastChange,omitzero"`336	// One entry per worker, from the reporter: the Job that ran337	// last for that worker and when it finished.338	Runs []libraryRun `json:"runs,omitempty"`339	// Gaps is one count per fact of the rows that fact has left to fill.340	// Waiting is the titles whose identity ended in candidates for a person to341	// choose from, and Unresolved the titles no provider could name. All three342	// are the reporter's own numbers.343	Gaps       map[string]int `json:"gaps,omitempty"`344	Waiting    int            `json:"waiting"`345	Unresolved int            `json:"unresolved"`346	// The titles a fact left because another writer holds the element group it347	// writes. The repair is to stop that writer for this library.348	Fights int `json:"fights"`349	// Webhook is the URL of this Library's webhook endpoint on the350	// operator, the address a person gives to Radarr, Sonarr, or351	// Jellyfin.352	Webhook    string      `json:"webhook,omitempty"`353	Conditions []Condition `json:"conditions,omitempty"`354}355356// LibraryVolume is the PersistentVolume the claim is bound to,357// reported here so that whoever plays a title from this library reads358// the volume without a second request for the claim. Type is the name of the volume's359// source key, and the NFS pair is filled only for an NFS volume.360type LibraryVolume struct {361	Name   string `json:"name"`362	Type   string `json:"type,omitempty"`363	Server string `json:"server,omitempty"`364	Path   string `json:"path,omitempty"`365}366367// A Player is media-operator's unit of equipment. This operator reads368// one for its status.idle alone: the block media-operator publishes for the controller it369// delegated the idle screen to. Nothing here is written back, so the type370// carries no spec.371type Player struct {372	APIVersion string       `json:"apiVersion,omitempty"`373	Kind       string       `json:"kind,omitempty"`374	Metadata   ObjectMeta   `json:"metadata"`375	Status     PlayerStatus `json:"status"`376}377378// The idle block the Player carries, and an empty one for a Player that379// carries none. The empty block names no controller and no claim, so a Player380// with no idle status is a Player this operator stands nothing for.381func (p *Player) idle() PlayerIdleStatus {382	if p.Status.Idle == nil {383		return PlayerIdleStatus{}384	}385	return *p.Status.Idle386}387388// Whether this operator stands the Player's idle screen. The pass acts389// on status.idle and never on spec.idle: the spec can inherit its controller390// from MediaPreferences, and media-operator alone resolves those tiers.391func (p *Player) delegated() bool {392	return p.idle().Controller == screenController393}394395// The collection ListPlayers answers. Its resourceVersion is where the396// player watch begins.397type PlayerList struct {398	Metadata ListMeta `json:"metadata"`399	Items    []Player `json:"items"`400}401402// The one name a MediaPreferences may take. media-operator's CRD pins it,403// and this operator reads the singleton by this name.404const mediaPreferencesName = "default"405406// The household defaults media-operator owns, read here for one field:407// the wall-clock zone every screen shows. Nothing is written back, so408// the type carries that field alone.409type MediaPreferences struct {410	APIVersion string               `json:"apiVersion,omitempty"`411	Kind       string               `json:"kind,omitempty"`412	Metadata   ObjectMeta           `json:"metadata"`413	Spec       MediaPreferencesSpec `json:"spec"`414}415416// The one field of the household defaults this operator reads.417type MediaPreferencesSpec struct {418	// The household wall-clock zone, an IANA name like America/New_York.419	// The browser pod reads it as TZ, so its clock and its day's draw420	// follow the house and not UTC.421	TimeZone string `json:"timeZone,omitempty"`422}423424// The collection ListMediaPreferences answers. Its resourceVersion is425// where the watch begins.426type MediaPreferencesList struct {427	Metadata ListMeta           `json:"metadata"`428	Items    []MediaPreferences `json:"items"`429}430431// The household zone the list holds: the default MediaPreferences' own,432// or nothing where the cluster states none. A pod with no TZ reads UTC,433// the way media-operator's own pods do.434func householdZone(list *MediaPreferencesList) string {435	for _, preferences := range list.Items {436		if preferences.Metadata.Name == mediaPreferencesName {437			return preferences.Spec.TimeZone438		}439	}440	return ""441}442443// The half of a Player's status this operator acts on. The idle block444// is absent on a Player that stands no idle screen.445type PlayerStatus struct {446	Idle *PlayerIdleStatus `json:"idle,omitempty"`447}448449// What media-operator publishes for the idle controller it delegated450// to. Controller names that controller, and this operator acts only on its own451// name. Claim is the ResourceClaim media-operator stood for the screen, in the452// Player's namespace, and Requests names the requests in it the browser453// container states. The requests are media-operator's own list: render is454// there only for a Player whose display claim holds one.455//456// FadeAfterSeconds and OffAfterSeconds are the seconds before the457// screen fades and the seconds before the panel goes dark.458// media-operator resolves both and always writes them, because zero is459// a policy and an absent field is not one. The browser holds the460// timers, through the media-screen crate, so media-operator settles the461// policy and the client runs it.462type PlayerIdleStatus struct {463	Controller string   `json:"controller"`464	Claim      string   `json:"claim,omitempty"`465	Requests   []string `json:"requests,omitempty"`466467	FadeAfterSeconds int64 `json:"fadeAfterSeconds"`468	OffAfterSeconds  int64 `json:"offAfterSeconds"`469470	Bus *PlayerIdleBus `json:"bus,omitempty"`471}472473// PlayerIdleBus is the broker and every topic a delegate's client reads474// or writes: the retained status, the level, the commands topic that475// carries the re-present, the panel topic the client states the panel476// desire on, and the unit's controllers. VolumeTopic is empty for a477// unit with no sinks, which is the speaker gate. An older478// media-operator publishes no block, and the browser then takes the479// keyboard alone.480type PlayerIdleBus struct {481	Address       string             `json:"address"`482	StatusTopic   string             `json:"statusTopic"`483	VolumeTopic   string             `json:"volumeTopic,omitempty"`484	CommandsTopic string             `json:"commandsTopic"`485	PanelTopic    string             `json:"panelTopic"`486	Remotes       []PlayerIdleRemote `json:"remotes,omitempty"`487}488489// PlayerIdleRemote is one of the unit's controllers as a client reads490// it: the topic its presses arrive on and the topic its focus mark is491// on. The list is in spec.remotes order, because that position is the492// index a focus moment carries.493type PlayerIdleRemote struct {494	Events string `json:"events"`495	Focus  string `json:"focus"`496}497498// Play is media-operator's unit of playback: the Players it plays on499// and the items it plays in order. This operator creates Plays and500// reads none, so the type carries no status.501type Play struct {502	APIVersion string     `json:"apiVersion,omitempty"`503	Kind       string     `json:"kind,omitempty"`504	Metadata   ObjectMeta `json:"metadata"`505	Spec       PlaySpec   `json:"spec"`506}507508// PlaySpec names the Players and the items. A request from one screen509// names one Player.510type PlaySpec struct {511	Players []string   `json:"players,omitempty"`512	Items   []PlayItem `json:"items,omitempty"`513}514515// PlayItem is one item: the media reference the Player accepts, and516// the words the film's own display shows. The reference is a claim517// URI, which mounts the library's claim read-only on the playback pod.518type PlayItem struct {519	URI          string            `json:"uri"`520	Presentation *PlayPresentation `json:"presentation,omitempty"`521}522523// PlayPresentation is what the display shows about one item, in524// media-operator's own field names. Every field is optional, because525// the catalog holds what the volume holds and no more.526type PlayPresentation struct {527	Type         string `json:"type,omitempty"`528	Hint         string `json:"hint,omitempty"`529	Title        string `json:"title,omitempty"`530	Series       string `json:"series,omitempty"`531	Season       int    `json:"season,omitempty"`532	Episode      int    `json:"episode,omitempty"`533	EpisodeTitle string `json:"episodeTitle,omitempty"`534	Year         int    `json:"year,omitempty"`535	Date         string `json:"date,omitempty"`536	Art          string `json:"art,omitempty"`537	Trickplay    string `json:"trickplay,omitempty"`538}539540// The condition types this operator publishes. Bound reports the541// storage and Ready reports the scanner. A library whose claim never542// binds shows the cause in Bound, and Ready reads NotBound beside it.543// Departing reports the teardown of a deleted Library: it is the one544// condition here whose True states work in progress rather than a545// healthy fact, and its reason says how far the teardown reached.546const (547	conditionBound     = "Bound"548	conditionReady     = "Ready"549	conditionDeparting = "Departing"550)551552// The reasons each condition takes. A reason is one CamelCase word a553// program matches on, and the message beside it is the sentence.554const (555	reasonBound          = "Bound"556	reasonClaimNotFound  = "ClaimNotFound"557	reasonClaimUnbound   = "ClaimUnbound"558	reasonVolumeNotFound = "VolumeNotFound"559560	reasonReady        = "Ready"561	reasonNotBound     = "NotBound"562	reasonNoCatalog    = "NoCatalog"563	reasonManyCatalogs = "ManyCatalogs"564	reasonNoReport     = "NoReport"565	// The namespace's catalog pod is not up, the schedule does566	// not stand yet, and the namespace's reporter has left the bus.567	reasonCatalogPending = "CatalogPending"568	reasonScanPending    = "ScanPending"569	reasonOffline        = "Offline"570571	// The Departing condition's reasons, in the order depart.go572	// reaches them: a scan Job of this library is still running, the573	// cleanup Job is deleting the rows, the Job finished and the574	// reporter has not echoed it yet, and the cleanup cannot run at575	// all. Blocked covers a cleanup Job that failed and a namespace576	// with more than one Catalog.577	reasonScanRunning  = "ScanRunning"578	reasonSweeping     = "Sweeping"579	reasonAwaitingEcho = "AwaitingEcho"580	reasonBlocked      = "Blocked"581	// An enricher Job of this library is still running. It writes onto the582	// volume and into the catalog the sweep is emptying.583	reasonEnrichRunning = "EnrichRunning"584)585586// The values status.phase takes. libraryPhase in status.go derives587// the first four from the Ready condition, the scanner's588// availability on the bus, and the newest report. Departing is not589// derived: depart.go writes it on a Library with a deletion590// timestamp, for as long as the finalizer holds the object open.591const (592	phasePending   = "Pending"593	phaseOffline   = "Offline"594	phaseScanning  = "Scanning"595	phaseEnriching = "Enriching"596	phaseIdle      = "Idle"597	phaseDeparting = "Departing"598	// Failed means the last scan of this library failed and wrote no599	// rows. The tables hold what the last good scan left.600	phaseFailed = "Failed"601)602603// ConditionStatus is a condition's verdict. It is a string rather than604// a bool because there is a third state: an operator must be able to605// say when it cannot tell yet.606type ConditionStatus string607608const (609	ConditionTrue    ConditionStatus = "True"610	ConditionFalse   ConditionStatus = "False"611	ConditionUnknown ConditionStatus = "Unknown"612)613614// Condition mirrors metav1.Condition, the shape Kubernetes uses615// everywhere, and liken's own. Anyone who reads kubectl describe616// output on a Pod already knows how to read one of these.617//618// ObservedGeneration records which metadata.generation the condition619// judged. Generation counts spec edits, so a reader can tell "Ready,620// for the spec as it stands" from "Ready, but for a spec two edits621// ago".622type Condition struct {623	Type               string          `json:"type"`624	Status             ConditionStatus `json:"status"`625	ObservedGeneration int64           `json:"observedGeneration,omitempty"`626	Reason             string          `json:"reason,omitempty"`627	Message            string          `json:"message,omitempty"`628	LastTransitionTime time.Time       `json:"lastTransitionTime"`629}630631// SetCondition adds or updates a condition by type. It keeps the632// Kubernetes rule that makes lastTransitionTime meaningful: the time633// moves only when Status flips, not on every write. That is what lets634// kubectl get answer "how long has this library been Ready?" instead635// of only "when did the operator last say so?".636func SetCondition(conditions []Condition, condition Condition, now time.Time) []Condition {637	condition.LastTransitionTime = now638	for i, existing := range conditions {639		if existing.Type != condition.Type {640			continue641		}642		if existing.Status == condition.Status {643			condition.LastTransitionTime = existing.LastTransitionTime644		}645		conditions[i] = condition646		return conditions647	}648	return append(conditions, condition)649}

1package main23// This is a Kubernetes client written straight against the HTTP API,4// following liken's own (kubernetes/apiclient.go) and the media5// operator's, for the same reason: the API is HTTPS that serves6// JSON, and client-go would bring informers, work queues, and7// generated types this program does not use.8//9// Every pod already holds what it needs to reach the API server.10// Kubernetes injects two environment variables that name the11// server's in-cluster address, and the kubelet mounts a CA12// certificate and a ServiceAccount token at a known path.1314import (15	"bytes"16	"context"17	"crypto/tls"18	"crypto/x509"19	"encoding/json"20	"errors"21	"fmt"22	"io"23	"net"24	"net/http"25	"os"26	"time"27)2829// The path the kubelet mounts the ServiceAccount credentials on.30const defaultServiceAccountDir = "/var/run/secrets/kubernetes.io/serviceaccount"3132// ServiceAccountDir is a variable so a test points it at a directory33// it controls.34var serviceAccountDir = defaultServiceAccountDir3536// These two answers are values, not failures. An absent object is the37// normal state the caller answers by creating it, and a conflict is38// the normal state under optimistic concurrency that the caller39// answers by reading again.40var (41	ErrNotFound = errors.New("not found")42	ErrConflict = errors.New("conflict: something else wrote this object first")43)4445type Client struct {46	base        string47	http        *http.Client48	credentials string49}5051// NewClient builds a client from its three parts. InClusterClient52// reads them from the pod's environment; a test hands in an53// httptest server's base and no credentials.54func NewClient(base string, httpClient *http.Client, credentials string) *Client {55	return &Client{base: base, http: httpClient, credentials: credentials}56}5758func InClusterClient() (*Client, error) {59	host, port := os.Getenv("KUBERNETES_SERVICE_HOST"), os.Getenv("KUBERNETES_SERVICE_PORT")60	if host == "" || port == "" {61		return nil, fmt.Errorf("not running in a cluster: KUBERNETES_SERVICE_HOST unset")62	}6364	// The client trusts the cluster's own CA and not the system65	// store, so it accepts this API server and no other server that66	// answers on the address.67	caPEM, err := os.ReadFile(serviceAccountDir + "/ca.crt")68	if err != nil {69		return nil, fmt.Errorf("reading service account CA: %w", err)70	}71	roots := x509.NewCertPool()72	if !roots.AppendCertsFromPEM(caPEM) {73		return nil, fmt.Errorf("service account CA contains no certificates")74	}7576	return NewClient("https://"+host+":"+port, &http.Client{77		Transport: &http.Transport{78			TLSClientConfig: &tls.Config{RootCAs: roots},79			// Each timeout bounds the same failure: a server that80			// stops answering without sending anything. There is no81			// overall client timeout, because a watch is a request82			// whose response never ends, and a whole-request deadline83			// would cut every stream on schedule.84			DialContext: (&net.Dialer{85				Timeout:   5 * time.Second,86				KeepAlive: 10 * time.Second,87			}).DialContext,88			ResponseHeaderTimeout: 10 * time.Second,89			IdleConnTimeout:       30 * time.Second,90		},91	}, serviceAccountDir), nil92}9394// RequestJSON sends one request and decodes the answer, turning every95// non-2xx status into an error that carries the server's own message.96func (c *Client) RequestJSON(ctx context.Context, method, path string, body []byte, out any) error {97	return c.RequestWithType(ctx, method, path, jsonContentType, body, out)98}99100// The two content types this client sends. A PATCH needs its own,101// because the API server reads which patch dialect a request speaks102// from the Content-Type header alone.103const (104	jsonContentType = "application/json"105	mergePatchType  = "application/merge-patch+json"106)107108// RequestWithType is RequestJSON with the request's own content type109// stated, which is what a merge patch needs.110func (c *Client) RequestWithType(ctx context.Context, method, path, contentType string, body []byte, out any) error {111	resp, err := c.do(ctx, method, path, contentType, body)112	if err != nil {113		return err114	}115	defer drain(resp.Body)116117	if resp.StatusCode == http.StatusNotFound {118		return ErrNotFound119	}120	if resp.StatusCode == http.StatusConflict {121		return ErrConflict122	}123	if resp.StatusCode < 200 || resp.StatusCode > 299 {124		message, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))125		return fmt.Errorf("%s %s: %s: %s", method, path, resp.Status, message)126	}127	if out == nil {128		return nil129	}130	return json.NewDecoder(resp.Body).Decode(out)131}132133// Do sends one request and hands back the open response, which is134// what a watch needs and what RequestJSON is built on.135//136// The context is the caller's, so a pass that ends takes its requests137// with it, and a watch runs for as long as its own context does.138func (c *Client) Do(ctx context.Context, method, path string, body []byte) (*http.Response, error) {139	return c.do(ctx, method, path, jsonContentType, body)140}141142// Do is Do with the body's content type stated, so one request path143// sends both a JSON write and a merge patch.144func (c *Client) do(ctx context.Context, method, path, contentType string, body []byte) (*http.Response, error) {145	var reader io.Reader146	if body != nil {147		reader = bytes.NewReader(body)148	}149	req, err := http.NewRequestWithContext(ctx, method, c.base+path, reader)150	if err != nil {151		return nil, err152	}153	// The token is read from disk on every request. The mounted154	// token is short-lived and the kubelet refreshes the file as155	// each one nears expiry, so a client that held one in memory156	// would start getting 401s.157	if c.credentials != "" {158		token, err := os.ReadFile(c.credentials + "/token")159		if err != nil {160			return nil, fmt.Errorf("reading service account token: %w", err)161		}162		req.Header.Set("Authorization", "Bearer "+string(token))163	}164	req.Header.Set("Accept", "application/json")165	if body != nil {166		req.Header.Set("Content-Type", contentType)167	}168	return c.http.Do(req)169}170171// Drain reads whatever the caller left in the body, then closes it.172// Go returns a connection to its pool only when the body reaches173// EOF, so an early close costs a fresh connection and TLS handshake,174// and reaches the server as a hang-up on a request it answered.175const maxDrain = 4 << 20176177func drain(body io.ReadCloser) {178	_, _ = io.Copy(io.Discard, io.LimitReader(body, maxDrain))179	_ = body.Close()180}181182// Every API server answers /version, and the answer needs no RBAC183// rule, so it is the cheapest proof that the client reached the184// server it was configured for.185const versionPath = "/version"186187// Version holds the one field of /version the operator reports.188type Version struct {189	GitVersion string `json:"gitVersion"`190}191192func ServerVersion(ctx context.Context, client *Client) (Version, error) {193	var version Version194	if err := client.RequestJSON(ctx, http.MethodGet, versionPath, nil, &version); err != nil {195		return Version{}, err196	}197	return version, nil198}199200// The collection paths. Libraries are listed and watched across every201// namespace and written back per namespace. The storage and the pods202// are ordinary core-group objects: a claim and a pod are namespaced,203// and a volume is not.204const (205	librariesPath = "/apis/" + libraryAPIVersion + "/libraries"206	// The Catalogs, listed and watched across every namespace and207	// written back per namespace, the same shape as the Libraries.208	catalogsPath = "/apis/" + libraryAPIVersion + "/catalogs"209	// The Players, listed and watched across every namespace,210	// read-only. A Player is media-operator's object, and this operator211	// reads the collection to find the screens delegated to it.212	playersPath = "/apis/" + playerAPIVersion + "/players"213	// The MediaPreferences, cluster-scoped and read-only here, for the214	// household zone the screen pods carry.215	mediaPreferencesPath = "/apis/" + playerAPIVersion + "/mediapreferences"216217	libraryPrefix = "/apis/" + libraryAPIVersion + "/namespaces/"218	corePrefix    = "/api/v1/namespaces/"219	volumesPath   = "/api/v1/persistentvolumes"220	podsAllPath   = "/api/v1/pods"221222	// The slices behind the catalog Services, one in every namespace that223	// holds a Library.224	endpointSlicePrefix = "/apis/" + endpointSliceAPIVersion + "/namespaces/"225)226227// CatalogMemberQuery narrows a pod list or a pod watch to the228// pods that hold a catalog agent, whatever kind of pod they are: the229// catalog pod, a running Job's pod, and a screen pod. The equals sign230// inside the selector is percent-encoded, so the server reads one231// parameter and not two.232const catalogMemberQuery = "labelSelector=" + memberLabelKey + "%3D" + memberLabelValue233234// The same narrowing for the screen pods, which carry a name label of235// their own. The two selectors keep the two kinds of pod apart, so a list of236// one never answers with the other.237const screenPodsQuery = "labelSelector=" + scannerLabelKey + "%3D" + screenLabelValue238239func libraryPath(namespace, name string) string {240	return libraryPrefix + namespace + "/libraries/" + name241}242243func catalogPath(namespace, name string) string {244	return libraryPrefix + namespace + "/catalogs/" + name245}246247func claimPath(namespace, name string) string {248	return corePrefix + namespace + "/persistentvolumeclaims/" + name249}250251func claimsPath(namespace string) string {252	return corePrefix + namespace + "/persistentvolumeclaims"253}254255// playsPath is the plays collection of one namespace. A Play is256// created in the Player's namespace, which is the Library's namespace257// as well, so no reference this operator makes crosses a namespace.258func playsPath(namespace string) string {259	return "/apis/" + playerAPIVersion + "/namespaces/" + namespace + "/plays"260}261262func podsPath(namespace string) string {263	return corePrefix + namespace + "/pods"264}265266func endpointSlicesPath(namespace string) string {267	return endpointSlicePrefix + namespace + "/endpointslices"268}269270func servicesPath(namespace string) string {271	return corePrefix + namespace + "/services"272}273274// ListLibraries answers a whole pass with one request, and the list's275// resourceVersion is where the libraries watch resumes from.276func ListLibraries(ctx context.Context, c *Client) (*LibraryList, error) {277	list := &LibraryList{}278	if err := c.RequestJSON(ctx, http.MethodGet, librariesPath, nil, list); err != nil {279		return nil, err280	}281	return list, nil282}283284// PutLibraryStatus writes through the status subresource, which is its285// own write path: this request can never touch a spec. The286// resourceVersion in the body is what makes the write conditional, so287// a status written over a Library that changed underneath answers288// ErrConflict and the next pass reads it again.289func PutLibraryStatus(ctx context.Context, c *Client, library *Library) (*Library, error) {290	body, err := json.Marshal(library)291	if err != nil {292		return nil, err293	}294	written := &Library{}295	path := libraryPath(library.Metadata.Namespace, library.Metadata.Name) + "/status"296	if err := c.RequestJSON(ctx, http.MethodPut, path, body, written); err != nil {297		return nil, err298	}299	return written, nil300}301302// PatchLibraryFinalizers writes a Library's finalizer list and303// answers with the resourceVersion the write produced, which a304// caller needs before it writes the same object again in one pass.305//306// It is a merge patch and not a replace, because a replace sends307// every field this program models and drops every field it does not,308// which would take a person's own labels and annotations off the309// Library. The resourceVersion inside the patch makes the write310// conditional the same way a replace is: a write that raced another311// answers ErrConflict instead of clobbering it.312func PatchLibraryFinalizers(ctx context.Context, c *Client, namespace, name, resourceVersion string, finalizers []string) (string, error) {313	body, err := json.Marshal(map[string]any{314		"metadata": map[string]any{315			"resourceVersion": resourceVersion,316			"finalizers":      finalizers,317		},318	})319	if err != nil {320		return "", err321	}322	var patched struct {323		Metadata ObjectMeta `json:"metadata"`324	}325	path := libraryPath(namespace, name)326	if err := c.RequestWithType(ctx, http.MethodPatch, path, mergePatchType, body, &patched); err != nil {327		return "", err328	}329	return patched.Metadata.ResourceVersion, nil330}331332// ListPlayers reads every Player in the cluster with one request, the333// way the pass reads the Libraries. The list's resourceVersion is where the334// player watch resumes from. A cluster with no media-operator serves no such335// collection, and the failure is the caller's to report and carry on from.336func ListPlayers(ctx context.Context, c *Client) (*PlayerList, error) {337	list := &PlayerList{}338	if err := c.RequestJSON(ctx, http.MethodGet, playersPath, nil, list); err != nil {339		return nil, err340	}341	return list, nil342}343344// ListMediaPreferences reads the household defaults with one request, on345// the same terms as the Players: a cluster with no media-operator serves no346// such collection, and the failure is the caller's to report and carry on347// from.348func ListMediaPreferences(ctx context.Context, c *Client) (*MediaPreferencesList, error) {349	list := &MediaPreferencesList{}350	if err := c.RequestJSON(ctx, http.MethodGet, mediaPreferencesPath, nil, list); err != nil {351		return nil, err352	}353	return list, nil354}355356// ListCatalogs answers a whole pass with one request, and the list's357// resourceVersion is where the catalogs watch resumes from.358func ListCatalogs(ctx context.Context, c *Client) (*CatalogList, error) {359	list := &CatalogList{}360	if err := c.RequestJSON(ctx, http.MethodGet, catalogsPath, nil, list); err != nil {361		return nil, err362	}363	return list, nil364}365366// PutCatalogStatus writes through the status subresource, so this367// request can never touch a spec. The resourceVersion in the body makes368// the write conditional, the same as PutLibraryStatus.369func PutCatalogStatus(ctx context.Context, c *Client, catalog *NamespaceCatalog) (*NamespaceCatalog, error) {370	body, err := json.Marshal(catalog)371	if err != nil {372		return nil, err373	}374	written := &NamespaceCatalog{}375	path := catalogPath(catalog.Metadata.Namespace, catalog.Metadata.Name) + "/status"376	if err := c.RequestJSON(ctx, http.MethodPut, path, body, written); err != nil {377		return nil, err378	}379	return written, nil380}381382// GetPersistentVolumeClaim reads the claim a Library names, for two383// answers: whether it is bound, and which volume it is bound to. An384// absent claim is ErrNotFound, which the pass reports as the385// ClaimNotFound reason rather than as a failure. It also reads the386// catalog claim the operator provisions, to tell an existing one from387// none.388func GetPersistentVolumeClaim(ctx context.Context, c *Client, namespace, name string) (*PersistentVolumeClaim, error) {389	claim := &PersistentVolumeClaim{}390	if err := c.RequestJSON(ctx, http.MethodGet, claimPath(namespace, name), nil, claim); err != nil {391		return nil, err392	}393	return claim, nil394}395396// CreatePersistentVolumeClaim provisions a catalog claim: a Library's,397// the catalog pod's, or a screen's. The operator creates one once and never398// updates it, because a claim's spec is immutable once it binds.399func CreatePersistentVolumeClaim(ctx context.Context, c *Client, claim *PersistentVolumeClaim) (*PersistentVolumeClaim, error) {400	body, err := json.Marshal(claim)401	if err != nil {402		return nil, err403	}404	created := &PersistentVolumeClaim{}405	if err := c.RequestJSON(ctx, http.MethodPost, claimsPath(claim.Metadata.Namespace), body, created); err != nil {406		return nil, err407	}408	return created, nil409}410411// DeletePersistentVolumeClaim removes the catalog claim of a screen412// the scheduler cannot place. It is the one claim this operator deletes. An413// already-absent claim is success, because the operator deletes the claim to414// replace it and a delete that races another pass must not fail.415func DeletePersistentVolumeClaim(ctx context.Context, c *Client, namespace, name string) error {416	err := c.RequestJSON(ctx, http.MethodDelete, claimPath(namespace, name), nil, nil)417	if errors.Is(err, ErrNotFound) {418		return nil419	}420	return err421}422423// GetPersistentVolume reads the volume behind a bound claim, for what424// serves it. A PersistentVolume is cluster-scoped, so the path carries425// no namespace.426func GetPersistentVolume(ctx context.Context, c *Client, name string) (*PersistentVolume, error) {427	volume := &PersistentVolume{}428	if err := c.RequestJSON(ctx, http.MethodGet, volumesPath+"/"+name, nil, volume); err != nil {429		return nil, err430	}431	return volume, nil432}433434// ListCatalogMemberPods reads every pod that holds a catalog435// agent across every namespace, because a Catalog is in whatever436// namespace its Libraries do. The list's resourceVersion is where the437// pod watch begins.438func ListCatalogMemberPods(ctx context.Context, c *Client) (*PodList, error) {439	list := &PodList{}440	if err := c.RequestJSON(ctx, http.MethodGet, podsAllPath+"?"+catalogMemberQuery, nil, list); err != nil {441		return nil, err442	}443	return list, nil444}445446// ListScreenPods reads this operator's screen pods across every447// namespace, on the same terms, so the pass knows which screens stand448// before it deletes one. They reach the catalog EndpointSlice through449// the member label like every other agent.450func ListScreenPods(ctx context.Context, c *Client) (*PodList, error) {451	list := &PodList{}452	if err := c.RequestJSON(ctx, http.MethodGet, podsAllPath+"?"+screenPodsQuery, nil, list); err != nil {453		return nil, err454	}455	return list, nil456}457458func GetPod(ctx context.Context, c *Client, namespace, name string) (*Pod, error) {459	pod := &Pod{}460	if err := c.RequestJSON(ctx, http.MethodGet, podsPath(namespace)+"/"+name, nil, pod); err != nil {461		return nil, err462	}463	return pod, nil464}465466func CreatePod(ctx context.Context, c *Client, pod *Pod) (*Pod, error) {467	body, err := json.Marshal(pod)468	if err != nil {469		return nil, err470	}471	created := &Pod{}472	if err := c.RequestJSON(ctx, http.MethodPost, podsPath(pod.Metadata.Namespace), body, created); err != nil {473		return nil, err474	}475	return created, nil476}477478// CreatePlay posts the Play one play request became. The API server479// mints the name from the prefix, because a person may start the same480// title twice and each start is its own Play.481func CreatePlay(ctx context.Context, c *Client, play *Play) (*Play, error) {482	body, err := json.Marshal(play)483	if err != nil {484		return nil, err485	}486	created := &Play{}487	if err := c.RequestJSON(ctx, http.MethodPost, playsPath(play.Metadata.Namespace), body, created); err != nil {488		return nil, err489	}490	return created, nil491}492493// DeletePod removes one pod this operator stands. An494// already-absent pod is success, because the operator deletes a pod to495// replace it and a delete that races another pass must not fail.496func DeletePod(ctx context.Context, c *Client, namespace, name string) error {497	err := c.RequestJSON(ctx, http.MethodDelete, podsPath(namespace)+"/"+name, nil, nil)498	if errors.Is(err, ErrNotFound) {499		return nil500	}501	return err502}503504// GetService reads the live catalog Service of one namespace. The505// read answers the fields the operator compares, and it answers the506// resourceVersion and the addresses the API server assigned, which507// the update carries back unchanged.508func GetService(ctx context.Context, c *Client, namespace, name string) (*Service, error) {509	service := &Service{}510	if err := c.RequestJSON(ctx, http.MethodGet, servicesPath(namespace)+"/"+name, nil, service); err != nil {511		return nil, err512	}513	return service, nil514}515516func CreateService(ctx context.Context, c *Client, service *Service) (*Service, error) {517	body, err := json.Marshal(service)518	if err != nil {519		return nil, err520	}521	created := &Service{}522	path := servicesPath(service.Metadata.Namespace)523	if err := c.RequestJSON(ctx, http.MethodPost, path, body, created); err != nil {524		return nil, err525	}526	return created, nil527}528529// UpdateService writes the whole Service back. The resourceVersion in530// the body makes the write conditional, so a Service that changed531// underneath answers ErrConflict, and the next pass reads it again.532func UpdateService(ctx context.Context, c *Client, service *Service) (*Service, error) {533	body, err := json.Marshal(service)534	if err != nil {535		return nil, err536	}537	written := &Service{}538	path := servicesPath(service.Metadata.Namespace) + "/" + service.Metadata.Name539	if err := c.RequestJSON(ctx, http.MethodPut, path, body, written); err != nil {540		return nil, err541	}542	return written, nil543}

1package main23// arrival.go is the arrival ledger as the walk reads it: the time a video4// file arrived, kept in .liken/arrival.yaml beside the files by the arrival5// fact in arrivalfact.go. The time is the inode change time and never the6// modification time, because an importer rewrites the modification time to7// the release date, and user space cannot set the change time, so the8// importer's rewrite stamps it with the real moment of import. The walk9// reads and never writes, because the scan Job mounts the volume read-only.10// This file holds the ledger's shape, its read, the per-folder read the11// walk makes, the change time, and the earliest-of fold.1213import (14	"errors"15	"os"16	"path/filepath"17	"syscall"18	"time"19)2021// The ledger file's name is the arrival fact's own, likenLedgerName of22// factArrival, so the fact's entries and its attempts are one file with one23// writer.24const arrivalLedgerName = "arrival.yaml"2526// The ledger's shape: one entry per video file the folder holds.27type arrivalLedger struct {28	Files []arrivalEntry `yaml:"files"`29}3031// One entry: the file's path relative to the folder, and the time the file32// arrived, RFC 3339 in UTC.33type arrivalEntry struct {34	Path string    `yaml:"path"`35	At   time.Time `yaml:"at"`36}3738// The walk's read of the entries alone, through the fact ledger reader, so39// the attempts beside them are read by the .liken pass and not here. A40// folder with no ledger reads as an empty ledger.41func readArrivalLedger(folder string) (arrivalLedger, error) {42	ledger, err := readLikenLedger(folder, factArrival)43	if err != nil {44		return arrivalLedger{}, err45	}46	return arrivalLedger{Files: ledger.Files}, nil47}4849// What the walk reads for one video. added is the ledger's time, or the50// change time where the ledger holds none. arrived is the ledger's time51// alone, and zero where it holds none, which is what the arrival fact's gap52// reads.53type fileArrival struct {54	added   int6455	arrived int6456}5758// The arrival of each video in a folder, read and never written. A ledger59// that cannot be read is an error the caller marks the pass incomplete60// with, and the change times stand for that pass. The change time is read61// only for a file with no entry, so a re-walk of a settled folder stats62// nothing here.63func folderArrivals(dir string, videos []string) (map[string]fileArrival, error) {64	ledger, readErr := readArrivalLedger(dir)65	held := map[string]int64{}66	for _, entry := range ledger.Files {67		held[entry.Path] = entry.At.Unix()68	}69	arrivals := map[string]fileArrival{}70	var statErr error71	for _, video := range videos {72		if at, known := held[video]; known {73			arrivals[video] = fileArrival{added: at, arrived: at}74			continue75		}76		at, err := changeTime(filepath.Join(dir, video))77		if err != nil {78			statErr = errors.Join(statErr, err)79			continue80		}81		arrivals[video] = fileArrival{added: at}82	}83	if readErr != nil {84		return arrivals, readErr85	}86	return arrivals, statErr87}8889// The inode change time, which is the real moment of import. User space90// cannot set it, so an importer that rewrites the modification time leaves it91// alone. This is the Linux stat, which is where the scanner runs.92func changeTime(path string) (int64, error) {93	var stat syscall.Stat_t94	if err := syscall.Stat(path, &stat); err != nil {95		return 0, &os.PathError{Op: "stat", Path: path, Err: err}96	}97	return stat.Ctim.Sec, nil98}99100// The earlier of two arrivals, where zero means none is known. A series or a101// set takes the first arrival among its members, and a member with none does102// not pull it to zero.103func earliestArrival(held, candidate int64) int64 {104	if candidate == 0 {105		return held106	}107	if held == 0 || candidate < held {108		return candidate109	}110	return held111}

1package main23// arrivalfact.go is the arrival fact: the enricher concern that writes the4// arrival ledger. It is a fact and not part of the walk because the scan Job5// mounts the volume read-only and the enrich Jobs mount it read-write. It6// asks no provider, because the file's own change time is the answer. It7// never rewrites an entry that exists, because the ledger is what makes the8// first sighting durable against every later sweep of the volume.910import (11	"context"12	"path/filepath"13	"time"14)1516// The name of the container that runs this fact.17const arrivalContainerName = "arrival"1819// The gap: a present video file whose files.arrived is zero, which is what20// the walk writes for a file the ledger holds no entry for, outside the21// attempt window. The fact's own row write and the next walk both close it.22func arrivalGapSQL() string {23	return `SELECT path FROM files ` +24		`WHERE library = ?1 AND type = '` + fileTypeVideo + `' AND present = 1 ` +25		`AND ` + gapClause(factArrival, "path", `arrived = 0`)26}2728// One folder's work: the folder whose .liken directory holds the ledger, and29// the entry path of each gap file under it, in the gap's order.30type arrivalWork struct {31	folder  string32	entries []string33}3435// The whole run. A catalog read that fails ends the container, because the36// gap list is the work. The gap is grouped by folder, because one folder is37// one file on the volume and one write, however many videos it holds.38func (e *enricher) arrivalFact(ctx context.Context) error {39	paths, err := e.gaps(ctx, factArrival, time.Now().UTC())40	if err != nil {41		return err42	}43	stamped := 044	for _, work := range e.arrivalWork(paths) {45		if err := ctx.Err(); err != nil {46			return err47		}48		stamped += e.stampArrivals(ctx, work)49	}50	e.logf("stamped %d of the %d files with no arrival", stamped, len(paths))51	return nil52}5354// The gap grouped by the folder that holds each file's ledger, in the order55// the gap named the folders. The files this Job's scope does not cover are56// left out.57func (e *enricher) arrivalWork(paths []string) []arrivalWork {58	var works []arrivalWork59	index := map[string]int{}60	for _, path := range paths {61		if !e.inScope(path) {62			continue63		}64		folder, entry := likenFolderFor(e.kind, filepath.Join(e.root, path))65		at, held := index[folder]66		if !held {67			at = len(works)68			index[folder] = at69			works = append(works, arrivalWork{folder: folder})70		}71		works[at].entries = append(works[at].entries, entry)72	}73	return works74}7576// One folder: an entry with the change time for every gap file the ledger77// holds none for, an attempt per file, and one write of the file. An entry78// that exists is kept as it is. Then the rows: files.arrived and the79// folder's item rows, so added follows. A volume that refuses the write is80// an error attempt written straight to the catalog, because the ledger is81// the one file this fact may write and a refused volume cannot hold the82// attempt. That row stands until the next walk sweeps it, so a refused83// volume costs one try per walk and never one per run. The answer is how84// many files gained an entry.85func (e *enricher) stampArrivals(ctx context.Context, work arrivalWork) int {86	now := time.Now().UTC()87	stamped := 088	err := e.writer.updateLikenLedger(work.folder, factArrival, func(ledger *likenLedger) {89		held := map[string]bool{}90		for _, entry := range ledger.Files {91			held[entry.Path] = true92		}93		for _, entry := range work.entries {94			result := attemptFound95			if !held[entry] {96				at, err := changeTime(filepath.Join(work.folder, entry))97				if err != nil {98					e.logf("could not read the change time of %s: %v", entry, err)99					result = attemptError100				} else {101					ledger.Files = append(ledger.Files, arrivalEntry{Path: entry, At: time.Unix(at, 0).UTC()})102					held[entry] = true103					stamped++104				}105			}106			ledger.noteAttempt(likenAttempt{Path: entry, At: now, Result: result})107		}108	})109	if err != nil {110		e.logf("could not write the arrival ledger at %s: %v", relativePath(e.root, work.folder), err)111		e.writeArrivalErrors(ctx, work, now)112		return 0113	}114	e.writeRows(factArrival, work.folder, true)115	return stamped116}117118// The error attempt rows of a folder the volume refused, one per gap file,119// keyed the way the walk keys a file fact's attempt.120func (e *enricher) writeArrivalErrors(ctx context.Context, work arrivalWork, at time.Time) {121	if e.catalog == nil {122		return123	}124	var rows []attemptRow125	for _, entry := range work.entries {126		rows = append(rows, attemptRow{127			Library: e.library,128			Item:    relativePath(e.root, filepath.Join(work.folder, entry)),129			Fact:    factArrival,130			At:      at.Unix(),131			Result:  attemptError,132		})133	}134	if _, err := e.catalog.UpsertAttempts(ctx, rows); err != nil {135		e.logf("could not write the %s attempt rows of %s: %v", factArrival, relativePath(e.root, work.folder), err)136	}137}

1package main23// The seam between one art fact and the providers that can answer it. The art4// group follows one rule: art is a single value, so the first block in the5// Library's sources that answers with an image wins, and no merge follows,6// where the nfo facts of a set take a union.78import (9	"context"10)1112// One image a provider offers for one art fact: the address the fetch reads,13// the language of the text in the image, and the count of votes behind it.14// The votes are TMDb's score or Fanart.tv's likes, so they order one15// provider's list and are never read across two providers.16type artCandidate struct {17	URL      string18	Language string19	Votes    float6420}2122// One provider block, asked for one gap and one art fact. It answers the23// images it holds, and a provider that holds none for that title answers no24// image and no error. The download is the provider's own, so a file takes the25// same retry rule the provider's calls take.26type artAnswerer interface {27	providerBlock() string28	serves(fact string) bool29	candidates(ctx context.Context, fact string, gap artGap, title titleRef) ([]artCandidate, error)30	fetchFile(ctx context.Context, address string) ([]byte, error)31}3233// The answerers the art container can ask, in order.34type artLine struct {35	answerers []artAnswerer36}3738// The line is built in the order LIBRARY_SOURCES names the blocks, which is39// the Library's own spec.sources order, and the rule for who answers reads40// that order. A block this image has no answerer for yet, and a block whose41// key did not reach the container, are both skipped with no error.42func newArtLine(blocks []string, value func(string) string) *artLine {43	line := &artLine{}44	for _, block := range blocks {45		token := value(providerTokenVariable(block))46		switch {47		case block == providerBlockTMDb && token != "":48			line.answerers = append(line.answerers, newTMDbArtAnswerer(newTMDbClient(tmdbAPIBase, token)))49		case block == providerBlockFanart && token != "":50			line.answerers = append(line.answerers, fanartArtAnswerer{client: newFanartClient(fanartAPIBase, token)})51		case block == providerBlockTVmaze:52			line.answerers = append(line.answerers, newTVmazeArtAnswerer(newTVmazeClient(tvmazeAPIBase)))53		}54	}55	return line56}5758// A fact with no answerer left has nothing to ask, so the titles that remain59// keep their gaps for the next run.60func (l *artLine) live(fact string) bool {61	for _, one := range l.answerers {62		if one.serves(fact) {63			return true64		}65	}66	return false67}6869// One gap's ask: every answerer that serves the fact, in order, until one of70// them holds an image, and that answerer is the one the download and the71// ledger name. An error does not end the ask, because a provider that is down72// leaves the blocks behind it their answer; the error is reported only where73// no block answered at all.74func (l *artLine) ask(ctx context.Context, fact string, gap artGap,75	title titleRef) (artAnswerer, []artCandidate, error) {76	var failure error77	for _, one := range l.answerers {78		if !one.serves(fact) {79			continue80		}81		candidates, err := one.candidates(ctx, fact, gap, title)82		if err != nil {83			if failure == nil {84				failure = err85			}86			continue87		}88		if len(candidates) > 0 {89			return one, candidates, nil90		}91	}92	return nil, nil, failure93}9495// The choice: the highest-voted image in the library's own language, then the96// highest-voted image with no language, then the highest-voted image of any97// language. Art with the title's own text is what a person reads on the98// screen, and art with no text reads in every language.99func chooseArt(candidates []artCandidate, language string) (artCandidate, bool) {100	for _, want := range []string{language, ""} {101		if candidate, held := bestArt(candidates, func(candidate artCandidate) bool {102			return candidate.Language == want103		}); held {104			return candidate, true105		}106	}107	return bestArt(candidates, func(artCandidate) bool { return true })108}109110// The highest-voted image the test admits. The first of two equal votes wins,111// which keeps the provider's own order.112func bestArt(candidates []artCandidate, admits func(artCandidate) bool) (artCandidate, bool) {113	best := artCandidate{}114	held := false115	for _, candidate := range candidates {116		if candidate.URL == "" || !admits(candidate) {117			continue118		}119		if !held || candidate.Votes > best.Votes {120			best, held = candidate, true121		}122	}123	return best, held124}

1package main23// The art phase's own table: the local name each fact writes, the TMDb list4// it reads, the size it fetches, and the gap query that says which files the5// library has none of. The fact names are in factnames.go, and the maps in6// enrich.go, factsrole.go, and attempts.go name these facts as they name7// every other.89import (10	"context"11	"fmt"12	"path/filepath"13	"strings"14)1516// The art facts this image can fill, in the order the art container names17// them in LIBRARY_FACTS: the title's own art first, then the season's, then18// the episode's.19var artFactNames = []string{20	factPoster, factBackdrop, factLogo, factClearart, factBanner,21	factLandscape, factDiscart, factSeasonPoster, factSeasonBanner, factEpisodeThumb,22}2324// The art facts the Library's own sources serve, in the order the group runs25// them. A Library whose sources hold one provider asks for what that provider26// serves and no more.27func servedArtFacts(library *Library, providers providerSet) []string {28	var served []string29	for _, fact := range artFactNames {30		if !artTypes[fact].holds(library.Spec.Kind) {31			continue32		}33		if providers.serving(library.Metadata.Namespace, library.Spec.Sources, fact) != nil {34			served = append(served, fact)35		}36	}37	return served38}3940// One art type. The file is the name Kodi and Jellyfin read beside the title,41// from https://kodi.wiki/view/Artwork and42// https://jellyfin.org/docs/general/server/media/movies/, both read on43// 2026-09-03. For the two season facts the file is the last part of the name44// alone, because the season number leads it, and the episode thumbnail carries45// no file at all, because it is named for its episode file. The kind is the46// one kind of library that carries this art, and empty for the art both kinds47// carry. The list is the TMDb array the fact reads, and the size is the one it48// fetches.49type artType struct {50	fact string51	file string52	kind string53	list string54	size string55}5657// Whether a library of this kind carries this art at all. A disc is a58// movie's, so a series is never a discart gap, and a series library never59// names the fact in its container.60func (t artType) holds(kind string) bool {61	return t.kind == "" || t.kind == kind62}6364// The ten types, with the sizes this project fetches from TMDb. The five65// names this wave adds come from the Kodi forum's artwork naming thread at66// https://forum.kodi.tv/showthread.php?tid=248825 and from67// https://kodi.wiki/view/Artwork/Season, both read on 2026-09-03. These sizes68// and not the original, because the browser of plan 22 draws on a 1080p panel69// and holds every decoded image in memory, and a raster over 2 MiB draws in70// bands. The sizes are the open decision plan 30 lists under what is not71// decided.72var artTypes = map[string]artType{73	factPoster:       {fact: factPoster, file: "poster.jpg", list: tmdbPosters, size: "w780"},74	factBackdrop:     {fact: factBackdrop, file: "fanart.jpg", list: tmdbBackdrops, size: "w1280"},75	factLogo:         {fact: factLogo, file: "clearlogo.png", list: tmdbLogos, size: "w500"},76	factClearart:     {fact: factClearart, file: "clearart.png"},77	factBanner:       {fact: factBanner, file: "banner.jpg"},78	factLandscape:    {fact: factLandscape, file: "landscape.jpg"},79	factDiscart:      {fact: factDiscart, file: "disc.png", kind: libraryKindMovies},80	factSeasonPoster: {fact: factSeasonPoster, file: "poster.jpg", list: tmdbPosters, size: "w780"},81	factSeasonBanner: {fact: factSeasonBanner, file: "banner.jpg"},82	factEpisodeThumb: {fact: factEpisodeThumb, list: tmdbStills, size: "w300"},83}8485// The start of the name Kodi reads for the art of season zero, which holds86// the specials.87const specialsSeasonPrefix = "season-specials-"8889// The name of one season's art. Kodi reads it in the series folder beside90// tvshow.nfo, not in the season folder.91func seasonArtName(season int, suffix string) string {92	if season == 0 {93		return specialsSeasonPrefix + suffix94	}95	return fmt.Sprintf("season%02d-%s", season, suffix)96}9798// The name of one episode's thumbnail: the episode file's own name with the99// extension replaced, which is what both players read.100func episodeThumbName(video string) string {101	base := filepath.Base(video)102	return strings.TrimSuffix(base, filepath.Ext(base)) + "-thumb.jpg"103}104105// The file one gap writes, relative to the folder that holds it. The facts of106// a title key on the file name itself, the two season facts key on the season107// number, and the episode thumbnail keys on the episode file it goes beside.108func (t artType) fileFor(gap artGap) string {109	switch t.fact {110	case factSeasonPoster, factSeasonBanner:111		return seasonArtName(gap.season, t.file)112	case factEpisodeThumb:113		return episodeThumbName(gap.key)114	default:115		return t.file116	}117}118119// How much memory the art container may take. It is above the scanner's120// because this container holds one image in memory while it writes it, where121// every other container holds one row at a time.122const artMemoryLimit = "256Mi"123124// The name of the container that runs the art facts.125const artContainerName = "art"126127// What the enricher records as the answer for a file another tool already128// wrote. The ledger says so, and the file is never opened.129const artProviderExisting = "existing"130131// The language the choice prefers. A Library declares no language yet, so132// this is the one the project's own libraries hold. The field belongs on the133// Library.134const artLanguage = "en"135136// One gap: the key the attempt is recorded under, the TMDb id of the title,137// and the season and episode numbers where the fact needs them. The key is138// the path of the file the fact writes, or the episode file the thumbnail139// goes beside, so a gap that is already filled leaves the list.140type artGap struct {141	key     string142	tmdb    string143	season  int144	episode int145}146147// The folder the file lands in and the entry the ledger keys on. Both come148// off the key, so one rule places the file, the ledger, and the attempt.149func (g artGap) folder() string {150	return filepath.Dir(g.key)151}152153func (g artGap) entry() string {154	return filepath.Base(g.key)155}156157// One fact's run, bound to its name, so every art fact runs the same loop158// over its own gap.159func artFactRun(fact string) factRun {160	return func(ctx context.Context, e *enricher) error { return e.artFact(ctx, fact) }161}162163// The gap query of the title's own art, over the movies and the series of one164// library. A gap is a title with a TMDb id and no file of that name in the165// catalog, outside the retry window. The query asks for the id because a gap166// the enricher cannot close would schedule a Job every pass for ever, so a167// title no provider can name is no gap. The join onto aliases is what reads168// the TMDb id of a series or a movie whose own id is under another scheme.169func titleArtGapSQL(fact string) string {170	art := artTypes[fact]171	branch := func(table, scope string) string {172		return `SELECT t.library AS library, t.path || '/` + art.file + `' AS file, ` +173			`substr(a.alias, length('` + scope + `:tmdb:') + 1) AS tmdb ` +174			`FROM ` + table + ` AS t JOIN aliases AS a ` +175			`ON a.library = t.library AND a.item = t.id AND a.alias LIKE '` + scope + `:tmdb:%'`176	}177	// The branches are the kinds this art belongs to, so the disc art reads178	// the movies and never the series.179	branches := []string{}180	if art.holds(libraryKindMovies) {181		branches = append(branches, branch("movies", scopeMovie))182	}183	if art.holds(libraryKindSeries) {184		branches = append(branches, branch("series", scopeSeries))185	}186	return `SELECT file, tmdb, 0, 0 FROM (` + strings.Join(branches, ` UNION ALL `) +187		`) AS wanted WHERE library = ?1 AND ` + gapClause(fact, "file", artFileClause())188}189190// The name of one season's art file, as a query builds it from the series191// path and the episode's season number. The gap query and the release date of192// a season both key on this name, so one expression writes it.193func seasonArtFileSQL(suffix string) string {194	return `s.path || '/' || CASE WHEN e.season = 0 THEN '` + specialsSeasonPrefix + suffix + `' ` +195		`ELSE printf('season%02d-` + suffix + `', e.season) END`196}197198// One row per season a library holds episodes of, because the catalog keeps199// no season item. The art lands in the series folder under Kodi's own name.200func seasonArtGapSQL(fact string) string {201	suffix := artTypes[fact].file202	return `SELECT file, tmdb, season, 0 FROM (` +203		`SELECT DISTINCT s.library AS library, ` +204		seasonArtFileSQL(suffix) + ` AS file, ` +205		`substr(a.alias, length('` + scopeSeries + `:tmdb:') + 1) AS tmdb, e.season AS season ` +206		`FROM episodes AS e ` +207		`JOIN series AS s ON s.library = e.library AND s.id = e.series ` +208		`JOIN aliases AS a ON a.library = s.library AND a.item = s.id ` +209		`AND a.alias LIKE '` + scopeSeries + `:tmdb:%'` +210		`) AS wanted WHERE library = ?1 AND ` + gapClause(fact, "file", artFileClause())211}212213// One row per episode with no image of its own. The episode keys on its own214// file, because the thumbnail is named for that file. The check reads the215// link table, so an image another tool named differently still counts as the216// episode's.217// The episodes that hold an image are one list built once from the library's218// files, never a subquery that reads the outer row, because that form walked219// every file of the library once per episode and took forty seconds.220func episodeThumbGapSQL() string {221	return `SELECT video, tmdb, season, episode FROM (` +222		`SELECT e.library AS library, e.path AS video, e.id AS item, ` +223		`substr(a.alias, length('` + scopeSeries + `:tmdb:') + 1) AS tmdb, ` +224		`e.season AS season, e.episode AS episode ` +225		`FROM episodes AS e ` +226		`JOIN series AS s ON s.library = e.library AND s.id = e.series ` +227		`JOIN aliases AS a ON a.library = s.library AND a.item = s.id ` +228		`AND a.alias LIKE '` + scopeSeries + `:tmdb:%'` +229		`) AS wanted WHERE library = ?1 AND ` + gapClause(factEpisodeThumb, "video",230		`item NOT IN (SELECT fi.item FROM file_items AS fi `+231			`JOIN files AS f ON f.library = fi.library AND f.path = fi.path `+232			`WHERE fi.library = ?1 `+233			`AND f.type = '`+fileTypeImage+`' `+234			`AND f.role IN ('`+fileRoleThumb+`', '`+fileRoleStill+`'))`)235}236237// Where each art fact's items carry their release date. A title's art reads238// the title's own, a season's art reads the earliest episode of that season,239// which is the day the season starts, and the episode thumbnail reads the240// episode. A season whose episodes carry no date at all reads as no date,241// because min ignores the null the empty column becomes.242func artReleaseDates(fact string) string {243	art := artTypes[fact]244	switch fact {245	case factEpisodeThumb:246		return `SELECT library, path AS item, released FROM episodes WHERE library = ?1`247	case factSeasonPoster, factSeasonBanner:248		return `SELECT e.library AS library, ` + seasonArtFileSQL(art.file) + ` AS item, ` +249			`min(nullif(e.released, '')) AS released FROM episodes AS e ` +250			`JOIN series AS s ON s.library = e.library AND s.id = e.series ` +251			`WHERE e.library = ?1 GROUP BY e.library, s.path, e.season`252	default:253		return titleReleaseDates(`path || '/` + art.file + `'`)254	}255}256257// The file the fact would write is not in the catalog. This is the whole of258// "written where none exists" as the catalog can answer it, and the container259// checks the volume itself before it writes. The library is the first bound260// parameter by number, never the derived table's own column, because a261// subquery that reads the outer row is run again for every title, and this262// one over a library's files took seven seconds a fact.263func artFileClause() string {264	return `file NOT IN (SELECT path FROM files WHERE files.library = ?1)`265}

1package main23// What the art facts ask TMDb: the image lists of a title, a season, and an4// episode, and the configuration that names the image host and the sizes.5// The answerer in tmdbart.go turns them into candidates, and the choice among6// those candidates is in artanswer.go. The download itself is the shared one in providerhttp.go, so a 429 from the7// image host waits the way a 429 from the API does.89import (10	"context"11	"fmt"12	"strconv"13	"strings"14)1516// The arrays TMDb answers images in. A movie and a series answer posters,17// backdrops, and logos, a season answers posters, and an episode answers18// stills.19const (20	tmdbPosters   = "posters"21	tmdbBackdrops = "backdrops"22	tmdbLogos     = "logos"23	tmdbStills    = "stills"24)2526// One image as TMDb states it. The language is null for an image with no text27// in it, which is why the choice below reads a null language as its second28// preference and never as a miss.29type tmdbImage struct {30	FilePath    string  `json:"file_path"`31	Language    string  `json:"iso_639_1"`32	VoteAverage float64 `json:"vote_average"`33	VoteCount   int     `json:"vote_count"`34}3536// One images answer, with every array the four endpoints return, so one type37// reads a movie, a series, a season, and an episode.38type tmdbImageAnswer struct {39	Posters   []tmdbImage `json:"posters"`40	Backdrops []tmdbImage `json:"backdrops"`41	Logos     []tmdbImage `json:"logos"`42	Stills    []tmdbImage `json:"stills"`43}4445func (a tmdbImageAnswer) list(name string) []tmdbImage {46	switch name {47	case tmdbPosters:48		return a.Posters49	case tmdbBackdrops:50		return a.Backdrops51	case tmdbLogos:52		return a.Logos53	default:54		return a.Stills55	}56}5758// What /configuration says about images: the host every file path hangs off,59// and the sizes each kind of image is served in. The enricher reads it once60// per fact and never asks for a size TMDb does not serve.61type tmdbConfiguration struct {62	Images struct {63		SecureBaseURL string   `json:"secure_base_url"`64		PosterSizes   []string `json:"poster_sizes"`65		BackdropSizes []string `json:"backdrop_sizes"`66		LogoSizes     []string `json:"logo_sizes"`67		StillSizes    []string `json:"still_sizes"`68	} `json:"images"`69}7071// The sizes of one list, as the configuration names them.72func (c tmdbConfiguration) sizes(list string) []string {73	switch list {74	case tmdbPosters:75		return c.Images.PosterSizes76	case tmdbBackdrops:77		return c.Images.BackdropSizes78	case tmdbLogos:79		return c.Images.LogoSizes80	default:81		return c.Images.StillSizes82	}83}8485// The size the fetch asks for: the type's own size where TMDb serves it, and86// the original where it does not, because every kind of image is served in87// the original.88const tmdbOriginalSize = "original"8990func (c tmdbConfiguration) sizeFor(art artType) string {91	for _, size := range c.sizes(art.list) {92		if size == art.size {93			return art.size94		}95	}96	return tmdbOriginalSize97}9899// The address of one image: the host, the size, and the path TMDb gave.100func (c tmdbConfiguration) imageURL(size, filePath string) string {101	return strings.TrimSuffix(c.Images.SecureBaseURL, "/") + "/" + size + filePath102}103104// The one read of the provider's own settings.105func (c *tmdbClient) configuration(ctx context.Context) (tmdbConfiguration, error) {106	var answer tmdbConfiguration107	if err := c.get(ctx, tmdbConfigurationPath, nil, &answer); err != nil {108		return tmdbConfiguration{}, err109	}110	if answer.Images.SecureBaseURL == "" {111		return tmdbConfiguration{}, fmt.Errorf("tmdb %s: the answer names no image host", tmdbConfigurationPath)112	}113	return answer, nil114}115116// The images of one gap, from the endpoint its fact reads: a movie, a series,117// one of its seasons, or one of its episodes.118func (c *tmdbClient) images(ctx context.Context, kind, fact string, gap artGap) (tmdbImageAnswer, error) {119	var answer tmdbImageAnswer120	if err := c.get(ctx, tmdbImagesPath(kind, fact, gap), nil, &answer); err != nil {121		return tmdbImageAnswer{}, err122	}123	return answer, nil124}125126// Which endpoint one fact reads. The season and the episode hang off the127// series, and the title's own art follows the kind of the library.128func tmdbImagesPath(kind, fact string, gap artGap) string {129	series := "/3/tv/" + gap.tmdb130	switch fact {131	case factSeasonPoster:132		return series + "/season/" + strconv.Itoa(gap.season) + "/images"133	case factEpisodeThumb:134		return series + "/season/" + strconv.Itoa(gap.season) +135			"/episode/" + strconv.Itoa(gap.episode) + "/images"136	}137	if kind == libraryKindSeries {138		return series + "/images"139	}140	return "/3/movie/" + gap.tmdb + "/images"141}

1package main23// The art container's run: what it reads out of the catalog for one art fact,4// how it answers a file that already exists, and what it writes where none5// does. An art fact never opens a file another tool wrote, so a poster6// Jellyfin wrote stays as it is, and the ledger records that the file was7// already there.89import (10	"context"11	"fmt"12	"os"13	"path/filepath"14	"time"15)1617// The line is built once for the container, so the settings one provider18// states are read once for every art fact the container runs. A container with19// no answerer at all is a manifest to repair, because the operator creates it20// only where a source serves one of its facts.21func (e *enricher) artFact(ctx context.Context, fact string) error {22	if e.art == nil {23		e.art = newArtLine(commaNames(os.Getenv(librarySourcesVariable)), os.Getenv)24	}25	if len(e.art.answerers) == 0 {26		return fmt.Errorf("no provider key reached this container, and the %s fact cannot ask without one", fact)27	}28	return e.artGap(ctx, fact, e.art)29}3031// A catalog read that fails ends the container, because the gap list is the32// work. A fact no answerer in the line serves reads no gap at all, so a33// Library whose sources hold one provider costs nothing for the art that34// provider does not serve. A provider that refuses one image records an error35// attempt, and the run carries on to the next.36func (e *enricher) artGap(ctx context.Context, fact string, line *artLine) error {37	if !line.live(fact) {38		return nil39	}40	gaps, err := e.catalog.artGaps(ctx, e.library, fact, time.Now().UTC(), e.refresh[fact])41	if err != nil {42		return err43	}44	written := 045	for _, gap := range gaps {46		if err := ctx.Err(); err != nil {47			return err48		}49		if !e.inScope(gap.key) {50			continue51		}52		if e.artOne(ctx, line, artTypes[fact], gap) {53			written++54		}55	}56	e.logf("wrote %d of the %d %s files the library had none of", written, len(gaps), fact)57	return nil58}5960// One gap. The volume is read before the provider is asked, because a file61// that landed since the last walk is the answer already and costs no call.62// Then one image is chosen, downloaded, and created, and the ledger records63// which provider answered.64func (e *enricher) artOne(ctx context.Context, line *artLine, art artType, gap artGap) bool {65	folder := filepath.Join(e.root, gap.folder())66	target := filepath.Join(folder, art.fileFor(gap))67	if held, err := fileExists(target); err != nil {68		e.logf("could not read %s: %v", target, err)69		e.recordArt(folder, art.fact, gap.entry(), "", attemptError)70		return false71	} else if held {72		e.recordArt(folder, art.fact, gap.entry(), artProviderExisting, attemptFound)73		return false74	}7576	answerer, candidates, err := line.ask(ctx, art.fact, gap, e.artTitle(gap))77	if err != nil {78		e.logf("could not read the %s of %s: %v", art.fact, gap.key, err)79		e.recordArt(folder, art.fact, gap.entry(), "", attemptError)80		return false81	}82	image, held := chooseArt(candidates, artLanguage)83	if answerer == nil || !held {84		e.logf("no provider holds the %s of %s", art.fact, gap.key)85		e.recordArt(folder, art.fact, gap.entry(), "", attemptNothing)86		return false87	}88	return e.writeArt(ctx, answerer, art, gap, folder, target, image)89}9091// The ids a fact asks with come off the sidecar itself, which is where the92// identity fact wrote every one of them. The art phase reads them because the93// gap carries the TMDb id alone and two of the three providers key on another94// id. A folder with no sidecar carries no id, which leaves those two providers95// no answer and is not an error.96func (e *enricher) artTitle(gap artGap) titleRef {97	sidecar, _ := identitySidecar(e.kind, filepath.Join(e.root, gap.folder()))98	document, err := os.ReadFile(sidecar)99	if err != nil {100		return titleRef{kind: e.kind}101	}102	return titleRef{kind: e.kind, ids: sidecarIDs(document)}103}104105// The download and the write. The bytes live from the answer to the rename106// and no longer. A create that finds the file there answers as the read above107// does, because another writer reached it first.108func (e *enricher) writeArt(ctx context.Context, answerer artAnswerer, art artType, gap artGap,109	folder, target string, image artCandidate) bool {110	data, err := answerer.fetchFile(ctx, image.URL)111	if err != nil {112		e.logf("could not read %s: %v", image.URL, err)113		e.recordArt(folder, art.fact, gap.entry(), "", attemptError)114		return false115	}116	written, err := e.writer.createOnce(target, data)117	if err != nil {118		e.logf("could not write %s: %v", target, err)119		e.recordArt(folder, art.fact, gap.entry(), "", attemptError)120		return false121	}122	if !written {123		e.recordArt(folder, art.fact, gap.entry(), artProviderExisting, attemptFound)124		return false125	}126	e.logf("wrote the %s of %s from %s", art.fact, gap.key, answerer.providerBlock())127	e.recordArt(folder, art.fact, gap.entry(), answerer.providerBlock(), attemptFound)128	return true129}130131// The item entry and the attempt are one write of one file, so a reader never132// sees an answer without its attempt. A provider name of nothing is a miss or133// an error, which the attempt itself states.134func (e *enricher) recordArt(folder, fact, entry, provider, result string) {135	now := time.Now().UTC()136	err := e.writer.updateLikenLedger(folder, fact, func(ledger *likenLedger) {137		if provider != "" {138			item := likenItem{Path: entry, Provider: providerNames{provider}}139			if provider != artProviderExisting {140				item.Written = now141			}142			ledger.noteItem(item)143		}144		ledger.noteAttempt(likenAttempt{Path: entry, At: now, Result: result})145	})146	if err != nil {147		e.logf("could not record the %s attempt at %s: %v", fact, entry, err)148	}149	e.writeRows(fact, folder, result == attemptFound)150}151152// One art fact's work list, out of the local copy of the catalog, with the153// same query the reporter counts the gap with. Every row names the file to154// write, the TMDb id to ask for, and the season and episode where the fact155// needs them.156func (c *Catalog) artGaps(ctx context.Context, library, fact string,157	now, refresh time.Time) ([]artGap, error) {158	var gaps []artGap159	err := c.stream(ctx, gapQueries[fact], gapParams(fact, library, now, refresh), func(cells []any) error {160		if len(cells) < 4 {161			return nil162		}163		key, _ := cells[0].(string)164		id, _ := cells[1].(string)165		if key == "" || id == "" {166			return nil167		}168		gaps = append(gaps, artGap{169			key:     key,170			tmdb:    id,171			season:  int(cellNumber(cells[2])),172			episode: int(cellNumber(cells[3])),173		})174		return nil175	})176	if err != nil {177		return nil, fmt.Errorf("reading the %s gap of %s: %w", fact, library, err)178	}179	return gaps, nil180}

1package main23// attempts.go is the attempts table's Go side: the row, the writes, and how4// the scanner lifts a folder's .liken files into the rows the gap queries5// read. The rows are derived from the volume, so a lost catalog gets them6// back on the next walk.78import (9	"context"10	"path/filepath"11	"strings"12	"time"13)1415// The column of the attempts table that holds the fact. Corrosion applies the16// difference between the schema file and the database on every start. It adds17// tables, columns, and indexes, and it refuses to remove a column. This18// column is also part of the primary key, and Corrosion refuses to add a19// primary key column to a table that exists. So the column keeps the name it20// was created with, every Go name is fact, and this constant is the one place21// the two meet.22const attemptFactColumn = "concern"2324// One enricher's last attempt at one item, as the attempts table holds it.25// For a file fact the item is the file's path under the library root.26type attemptRow struct {27	Library string28	Item    string29	Fact    string30	At      int6431	Result  string32	// The provider block that answered, empty for a fact that asks no provider.33	// A set fact joins the blocks it took the union of with commas.34	Provider string35}3637// One attempts row, by the two key columns that follow the library.38type attemptKey struct {39	Item string40	Fact string41}4243// A repeat write updates the row in place, because one item and one fact44// hold one attempt, the latest. The update names no key column, so the row's45// identity never moves.46func (c *Catalog) UpsertAttempts(ctx context.Context, rows []attemptRow) (int, error) {47	statements := make([]statement, len(rows))48	for i, row := range rows {49		statements[i] = statement{50			sql: `INSERT INTO attempts (library, item, ` + attemptFactColumn + `, at, result, provider) VALUES (?, ?, ?, ?, ?, ?) ` +51				`ON CONFLICT (library, item, ` + attemptFactColumn + `) DO UPDATE SET at = excluded.at, ` +52				`result = excluded.result, provider = excluded.provider`,53			params: []any{row.Library, row.Item, row.Fact, row.At, row.Result, row.Provider},54		}55	}56	return c.apply(ctx, statements)57}5859// The delete names all three key columns, as the link table's delete does, so60// a sweep takes exactly the rows it marked.61func (c *Catalog) DeleteAttempts(ctx context.Context, library string, keys []attemptKey) (int, error) {62	statements := make([]statement, len(keys))63	for i, key := range keys {64		statements[i] = statement{65			sql:    `DELETE FROM attempts WHERE library = ? AND item = ? AND ` + attemptFactColumn + ` = ?`,66			params: []any{library, key.Item, key.Fact},67		}68	}69	return c.apply(ctx, statements)70}7172// The two key columns travel as one string through a sweep, joined by a73// separator no path or id holds, so the sweep's mark table keeps one column74// for every table it covers.75func attemptKeys(keys []string) []attemptKey {76	out := make([]attemptKey, len(keys))77	for i, key := range keys {78		item, fact, _ := strings.Cut(key, linkKeySeparator)79		out[i] = attemptKey{Item: item, Fact: fact}80	}81	return out82}8384func attemptSeenKey(row attemptRow) string {85	return row.Item + linkKeySeparator + row.Fact86}8788// Reads the attempts this library holds that the current epoch did not mark,89// one bounded batch, with the two key columns joined the way the mark joined90// them.91func attemptPruneSQL() string {92	return `SELECT item || char(31) || ` + attemptFactColumn + ` FROM attempts` +93		` WHERE library = ?` +94		` AND '` + seenAttempt + `' || item || char(31) || ` + attemptFactColumn +95		` NOT IN (SELECT id FROM seen WHERE epoch = ?)` +96		` AND at < ?` +97		` LIMIT ?`98}99100// How a rescan reaches one folder's attempts: a file fact keys on a path101// under the folder, and an item fact keys on the id of an item the folder102// holds.103func scopedAttemptPruneSQL() string {104	scope := func(table string) string {105		return `SELECT id FROM ` + table + ` WHERE library = ? AND ` + pathScopeClause("path")106	}107	return `SELECT item || char(31) || ` + attemptFactColumn + ` FROM attempts` +108		` WHERE library = ?` +109		` AND '` + seenAttempt + `' || item || char(31) || ` + attemptFactColumn +110		` NOT IN (SELECT id FROM seen WHERE epoch = ?)` +111		` AND (` + pathScopeClause("item") +112		` OR item IN (` + scope("movies") + ` UNION ` + scope("series") + ` UNION ` + scope("episodes") + `))` +113		` AND at < ?` +114		` LIMIT ?`115}116117func scopedAttemptPruneParams(library, folder string, epoch int64) []any {118	params := []any{library, epoch}119	params = append(params, pathScopeParams(folder)...)120	for range 3 {121		params = append(params, library)122		params = append(params, pathScopeParams(folder)...)123	}124	return append(params, walkStart(epoch), pruneBatch)125}126127// What one folder's .liken directory means to the scanner: which item the128// folder's own entry names, and which item each file under it names.129type likenSidecar struct {130	root    string131	dir     string132	library string133	item    string134	items   map[string]string135	// The facts whose ledgers this folder can hold. A title folder holds the136	// title's own, which is the list below, and a person's directory holds the137	// three contributor facts and no other.138	facts []string139}140141// The facts this folder is read for, which is the title list where the caller142// names none.143func (s likenSidecar) ledgerFacts() []string {144	if s.facts != nil {145		return s.facts146	}147	return likenFacts148}149150// The facts the scanner lifts out of a folder. A file fact keys on a151// path, because it works per file, and the identity fact keys on an item152// id, because it works per title.153var likenFacts = []string{factProbe, factArrival, factTrickplay, factIdentity,154	factOverview, factCertification,155	factRatingTMDb, factRatingIMDb, factRatingRottenTomatoes, factRatingMetacritic,156	factCredits,157	factPoster, factBackdrop, factLogo, factClearart, factBanner,158	factLandscape, factDiscart, factSeasonPoster, factSeasonBanner, factEpisodeThumb}159160// Reads every .liken file the folder holds into attempts rows. A folder that161// holds none reads as no rows and not as an error, because most folders hold162// none.163// One pass answers for both kinds of row, because the credits ledger the164// credits fact wrote is one of the files this pass already opens.165func (s likenSidecar) read() ([]attemptRow, []creditRow, error) {166	var rows []attemptRow167	var credits []creditRow168	for _, fact := range s.ledgerFacts() {169		ledger, err := readLikenLedger(s.dir, fact)170		if err != nil {171			return rows, credits, err172		}173		if fact == factCredits {174			credits = append(credits, creditRows(s.library, s.item, ledger.Credits)...)175		}176		for _, attempt := range ledger.Attempts {177			item := s.itemOf(fact, attempt.Path)178			if item == "" || attempt.Result == "" {179				continue180			}181			rows = append(rows, attemptRow{182				Library:  s.library,183				Item:     item,184				Fact:     fact,185				At:       attempt.At.Unix(),186				Result:   attempt.Result,187				Provider: strings.Join(attempt.Provider, ","),188			})189		}190	}191	return rows, credits, nil192}193194// How an entry's path resolves: a file fact names the file itself, and an195// item fact names the title the folder holds.196func (s likenSidecar) itemOf(fact, path string) string {197	if _, art := artTypes[fact]; fact == factProbe || fact == factArrival || fact == factTrickplay || art {198		return relativePath(s.root, filepath.Join(s.dir, path))199	}200	if path == likenSelfPath || path == "" {201		return s.item202	}203	return s.items[path]204}205206// A folder whose .liken files cannot be read marks the pass incomplete, the207// way an unreadable sidecar does, so the sweep never removes rows the volume208// still holds.209func readLikenSidecar(sidecar likenSidecar, result *walkResult) {210	rows, credits, err := sidecar.read()211	result.noteReadError(err)212	result.attempts = append(result.attempts, rows...)213	result.credits = append(result.credits, credits...)214}215216// The reporter counts a gap with the same query the container works from, so217// the number the operator schedules on and the rows the container finds are218// one set.219//220// The reporter runs in the catalog pod, which reads no Library, so it221// counts with no refresh time and publishes the oldest attempt of each222// fact beside the counts. The operator holds the Library and reads the223// two together.224func (c *Catalog) gapCounts(ctx context.Context, library string, now time.Time) (map[string]int, error) {225	counts := map[string]int{}226	for fact, query := range gapQueries {227		count, err := c.queryInt(ctx, `SELECT count(*) FROM (`+query+`)`,228			gapParams(fact, library, now, time.Time{}))229		if err != nil {230			return nil, err231		}232		counts[fact] = count233	}234	return counts, nil235}236237// The oldest attempt one library holds for each fact, which is what238// says whether a refresh time has work left: an attempt older than the refresh239// is a title that fact asks about again.240// A fact with no attempt at all has no entry.241const oldestAttemptQuery = `SELECT ` + attemptFactColumn + `, min(at) FROM attempts ` +242	`WHERE library = ? GROUP BY ` + attemptFactColumn243244// The oldest attempt per fact, read with one statement, because a245// statement per fact is one round trip per fact on every report.246func (c *Catalog) oldestAttempts(ctx context.Context,247	library string) (map[string]time.Time, error) {248	oldest := map[string]time.Time{}249	err := c.stream(ctx, oldestAttemptQuery, []any{library}, func(cells []any) error {250		if len(cells) < 2 {251			return nil252		}253		fact, _ := cells[0].(string)254		if fact == "" {255			return nil256		}257		oldest[fact] = time.Unix(int64(cellNumber(cells[1])), 0).UTC()258		return nil259	})260	if err != nil {261		return nil, err262	}263	return oldest, nil264}265266// The fights of one library: the attempts that found an element group another267// writer had changed, over every fact. A person reads it on the Library, and268// the repair is to stop the other writer.269func (c *Catalog) fightCount(ctx context.Context, library string) (int, error) {270	return c.queryInt(ctx, fightsQuery, []any{library})271}272273// The two counts a person reads on the Library beside the gaps: the titles274// that wait for a person, and the titles no provider could name.275func (c *Catalog) identityCounts(ctx context.Context, library string) (int, int, error) {276	waiting, err := c.queryInt(ctx, waitingQuery, []any{library})277	if err != nil {278		return 0, 0, err279	}280	unresolved, err := c.queryInt(ctx, unresolvedQuery, []any{library})281	if err != nil {282		return 0, 0, err283	}284	return waiting, unresolved, nil285}

1package main23// One namespace's reporter is either on the bus or it is not,4// and that one signal stands for every Library of the namespace,5// because the catalog pod is the one process that reports them all.67import "sync"89// The desk that holds the last availability the bus carried for10// each namespace's reporter, keyed by namespace, with the loop's own11// wake beside it.12type reporters struct {13	mutex  sync.Mutex14	online map[string]bool15	wake   chan<- struct{}16}1718func newReporters(wake chan<- struct{}) *reporters {19	return &reporters{online: map[string]bool{}, wake: wake}20}2122// A change of the flag wakes the loop, so the pass that answers23// it reads the Libraries beside the reporter that just arrived or left;24// a repeat wakes nothing, because a reconnecting reporter republishes25// online on every session and the pass has already run.26func (r *reporters) mark(namespace string, online bool) {27	r.mutex.Lock()28	previous, had := r.online[namespace]29	r.online[namespace] = online30	r.mutex.Unlock()31	if !had || previous != online {32		poke(r.wake)33	}34}3536// A namespace the desk holds nothing for reads offline, which is37// the state before its catalog pod first connects.38func (r *reporters) onlineFor(namespace string) bool {39	r.mutex.Lock()40	defer r.mutex.Unlock()41	return r.online[namespace]42}

1package main23// The batch objects this operator writes and the requests it4// makes for them, hand-written in the same form as the core objects in5// objects.go and reached through the same client. Every worker of a6// namespace is a Job: a scan runs on a schedule from a CronJob, a7// folder scan runs once from a Job the webhook creates, and a departure8// runs once from a Job of its own.910import (11	"context"12	"encoding/json"13	"errors"14	"fmt"15	"net/http"16	"os"17	"time"18)1920// The group the Job and the CronJob belong to.21const batchAPIVersion = "batch/v1"2223// The label Kubernetes stamps on every pod a Job creates, whose24// value is the Job's own name; the scanner reads it through the25// downward API and writes it into the runs row.26const jobNameLabel = "batch.kubernetes.io/job-name"2728// The field path the downward API reads that label from.29const jobNameFieldPath = "metadata.labels['" + jobNameLabel + "']"3031// One run of one worker. The operator writes the spec and reads32// the status, which is the count of pods in each state.33type Job struct {34	APIVersion string     `json:"apiVersion,omitempty"`35	Kind       string     `json:"kind,omitempty"`36	Metadata   ObjectMeta `json:"metadata"`37	Spec       JobSpec    `json:"spec"`38	Status     JobStatus  `json:"status"`39}4041// The collection ListWorkerJobs answers.42type JobList struct {43	Metadata ListMeta `json:"metadata"`44	Items    []Job    `json:"items"`45}4647// BackoffLimit is how many times Kubernetes replaces a failed48// pod before the Job itself fails, and TTLSecondsAfterFinished is how49// long a finished Job stays for a person to read its logs.50type JobSpec struct {51	BackoffLimit            *int32          `json:"backoffLimit,omitempty"`52	TTLSecondsAfterFinished *int32          `json:"ttlSecondsAfterFinished,omitempty"`53	Template                PodTemplateSpec `json:"template"`54}5556// JobStatus is what the Job controller reports: the counts of pods in each57// state, the time it ended the Job, and the conditions, which are its verdict58// on the whole Job. The counts alone cannot say whether a Job is over. A Job59// between the pods of its backoff counts no active pod and is not over.60//61// The completion time is the controller's own stamp on a Job whose pod exited62// zero. The early delete below measures its grace from that stamp, not from63// the pod's exit, so the two clocks are the controller's.64type JobStatus struct {65	Active         int            `json:"active,omitempty"`66	Succeeded      int            `json:"succeeded,omitempty"`67	Failed         int            `json:"failed,omitempty"`68	CompletionTime time.Time      `json:"completionTime,omitzero"`69	Conditions     []JobCondition `json:"conditions,omitempty"`70}7172// JobCondition is one verdict of the Job controller, in the shape73// batch/v1 writes it.74type JobCondition struct {75	Type   string          `json:"type"`76	Status ConditionStatus `json:"status"`77	Reason string          `json:"reason,omitempty"`78}7980// The two condition types that end a Job, one for each way it ends.81const (82	jobComplete = "Complete"83	jobFailed   = "Failed"84)8586// A Job is still doing its work while it has a pod running.87func (j *Job) active() bool { return j.Status.Active > 0 }8889// finished is true when the controller has ended the Job, and not90// before. A Job that waits out its backoff with no pod is unfinished.91func (j *Job) finished() bool {92	return j.holds(jobComplete) || j.holds(jobFailed)93}9495// succeeded is true when the controller ended the Job on a pod that exited96// zero. That is the one outcome the operator deletes early. A failed Job97// stays for its TTL, because a person reads its logs.98func (j *Job) succeeded() bool { return j.holds(jobComplete) }99100// holds is true when the Job carries one condition of the given type with101// status True. That is how batch/v1 writes a verdict it stands behind. A102// condition with status False or Unknown is not a verdict.103func (j *Job) holds(conditionType string) bool {104	for _, condition := range j.Status.Conditions {105		if condition.Type == conditionType && condition.Status == ConditionTrue {106			return true107		}108	}109	return false110}111112// The pod one Job or CronJob creates, as the metadata and spec113// the controller stamps onto it.114type PodTemplateSpec struct {115	Metadata ObjectMeta `json:"metadata"`116	Spec     PodSpec    `json:"spec"`117}118119// The schedule one Library's full walk runs on. The operator120// writes it and reads nothing back but the resourceVersion, because the121// Jobs it creates are read through the Job list.122type CronJob struct {123	APIVersion string      `json:"apiVersion,omitempty"`124	Kind       string      `json:"kind,omitempty"`125	Metadata   ObjectMeta  `json:"metadata"`126	Spec       CronJobSpec `json:"spec"`127}128129// ConcurrencyPolicy is Forbid, so a walk that runs past its next130// turn skips that turn rather than starting a second walk on a claim131// that admits one writer.132type CronJobSpec struct {133	Schedule                   string          `json:"schedule"`134	ConcurrencyPolicy          string          `json:"concurrencyPolicy,omitempty"`135	SuccessfulJobsHistoryLimit *int32          `json:"successfulJobsHistoryLimit,omitempty"`136	FailedJobsHistoryLimit     *int32          `json:"failedJobsHistoryLimit,omitempty"`137	JobTemplate                JobTemplateSpec `json:"jobTemplate"`138}139140// The Job one turn of the schedule creates.141type JobTemplateSpec struct {142	Metadata ObjectMeta `json:"metadata"`143	Spec     JobSpec    `json:"spec"`144}145146// The batch collections: the Jobs of every namespace read with one147// request, and the Jobs and CronJobs of one namespace written per148// namespace.149const (150	jobsAllPath = "/apis/" + batchAPIVersion + "/jobs"151	batchPrefix = "/apis/" + batchAPIVersion + "/namespaces/"152)153154func jobsPath(namespace string) string {155	return batchPrefix + namespace + "/jobs"156}157158func cronJobsPath(namespace string) string {159	return batchPrefix + namespace + "/cronjobs"160}161162// The narrowing that keeps a Job list to this operator's own163// workers, by the name label every Job it creates carries. The equals164// sign is percent-encoded, so the server reads one parameter.165const workerJobsQuery = "labelSelector=" + scannerLabelKey + "%3D" + workerLabelValue166167// A delete of a Job removes the pods under it as well, which the168// default policy of orphaning would leave behind holding the claim.169const backgroundDeletion = "?propagationPolicy=Background"170171// ListWorkerJobs reads this operator's Jobs across every namespace,172// because a Library is in whatever namespace its claim is.173func ListWorkerJobs(ctx context.Context, c *Client) (*JobList, error) {174	list := &JobList{}175	if err := c.RequestJSON(ctx, http.MethodGet, jobsAllPath+"?"+workerJobsQuery, nil, list); err != nil {176		return nil, err177	}178	return list, nil179}180181func CreateJob(ctx context.Context, c *Client, job *Job) (*Job, error) {182	body, err := json.Marshal(job)183	if err != nil {184		return nil, err185	}186	created := &Job{}187	if err := c.RequestJSON(ctx, http.MethodPost, jobsPath(job.Metadata.Namespace), body, created); err != nil {188		return nil, err189	}190	return created, nil191}192193// DeleteJob removes one Job and the pods under it. An already-absent194// Job is success, the rule DeletePod follows.195func DeleteJob(ctx context.Context, c *Client, namespace, name string) error {196	path := jobsPath(namespace) + "/" + name + backgroundDeletion197	err := c.RequestJSON(ctx, http.MethodDelete, path, nil, nil)198	if errors.Is(err, ErrNotFound) {199		return nil200	}201	return err202}203204// How long a succeeded worker Job and its pod stay before the pass deletes205// them. The Job's own TTL is the hour a failed Job keeps for a person to206// read, and it is the backstop when the operator is down.207const succeededJobGrace = 5 * time.Minute208209// retireSucceededJobs deletes every worker Job that exited zero longer than210// the grace ago. The pass acts on the list it already read, so a Job it211// deletes still decides this pass and is gone from the next one. A Job of a212// webhook chain stays for its TTL, whatever its stage: the chain reads its213// stages out of the Job list, and a rescan deleted while its enrich Job214// stands would be created again every pass. A delete that fails is reported215// and the pass carries on, because the next pass reads the Job again.216func (o *operator) retireSucceededJobs(ctx context.Context, jobs []Job, now time.Time) {217	cutoff := now.Add(-succeededJobGrace)218	for index := range jobs {219		job := &jobs[index]220		if !job.succeeded() || job.Status.CompletionTime.IsZero() ||221			!job.Status.CompletionTime.Before(cutoff) {222			continue223		}224		if job.Metadata.Annotations[chainAnnotation] != "" {225			continue226		}227		if err := DeleteJob(ctx, o.client, job.Metadata.Namespace, job.Metadata.Name); err != nil {228			fmt.Fprintf(os.Stderr, "retiring the finished job %s/%s: %v\n",229				job.Metadata.Namespace, job.Metadata.Name, err)230		}231	}232}233234func GetCronJob(ctx context.Context, c *Client, namespace, name string) (*CronJob, error) {235	cronJob := &CronJob{}236	path := cronJobsPath(namespace) + "/" + name237	if err := c.RequestJSON(ctx, http.MethodGet, path, nil, cronJob); err != nil {238		return nil, err239	}240	return cronJob, nil241}242243func CreateCronJob(ctx context.Context, c *Client, cronJob *CronJob) (*CronJob, error) {244	body, err := json.Marshal(cronJob)245	if err != nil {246		return nil, err247	}248	created := &CronJob{}249	path := cronJobsPath(cronJob.Metadata.Namespace)250	if err := c.RequestJSON(ctx, http.MethodPost, path, body, created); err != nil {251		return nil, err252	}253	return created, nil254}255256// UpdateCronJob writes the whole CronJob back. The resourceVersion in257// the body makes the write conditional, so a CronJob that changed258// underneath answers ErrConflict and the next pass reads it again.259func UpdateCronJob(ctx context.Context, c *Client, cronJob *CronJob) (*CronJob, error) {260	body, err := json.Marshal(cronJob)261	if err != nil {262		return nil, err263	}264	written := &CronJob{}265	path := cronJobsPath(cronJob.Metadata.Namespace) + "/" + cronJob.Metadata.Name266	if err := c.RequestJSON(ctx, http.MethodPut, path, body, written); err != nil {267		return nil, err268	}269	return written, nil270}271272// DeleteCronJob removes one Library's schedule. An already-absent273// CronJob is success, the rule DeleteJob follows.274func DeleteCronJob(ctx context.Context, c *Client, namespace, name string) error {275	path := cronJobsPath(namespace) + "/" + name + backgroundDeletion276	err := c.RequestJSON(ctx, http.MethodDelete, path, nil, nil)277	if errors.Is(err, ErrNotFound) {278		return nil279	}280	return err281}

1package main23// The bus client is one live TCP connection to the broker, with a4// reader, a single writer, and a keepalive timer, and it reconnects5// with backoff whenever any of the three fails. Every mode of this6// operator that speaks to the broker holds one Bus and lets it manage7// the connection, so no caller writes the socket directly.8//9// All writes go through one goroutine and one channel, because two10// goroutines writing one TCP connection would interleave their bytes11// and corrupt a packet. A publish or a subscribe from any goroutine12// enqueues a finished frame, and the writer sends it.1314import (15	"bufio"16	"context"17	"fmt"18	"net"19	"os"20	"sort"21	"sync"22	"time"23)2425// The keepalive the client asks the broker for, and the queue the26// writer drains. The client sends a PINGREQ once the connection has27// been idle longer than half the keepalive, so the broker never28// reaches the keepalive without hearing from the client. The queue is29// bounded, and a publish that would overflow it is dropped, which is30// correct at QoS 0.31const (32	busKeepalive  = 3033	busQueueDepth = 6434)3536// The reconnect backoff bounds. The client waits busMinBackoff after37// the first failure and doubles the wait up to busMaxBackoff, so a38// broker that is down does not become a tight reconnect loop. Both are39// variables so a test drives a reconnect in milliseconds.40var (41	busMinBackoff = time.Second42	busMaxBackoff = 30 * time.Second43)4445// busHandler receives one inbound message's topic and payload. The Bus46// calls it on the reader goroutine, so a handler that blocks holds up47// every later message on the connection.48type busHandler func(topic string, payload []byte)4950// busWill is the MQTT Last Will the client names at connect time. The51// broker publishes it on any disconnect the client does not make52// cleanly, which is how a killed pod's status is marked offline.53type busWill struct {54	Topic    string55	Payload  []byte56	Retained bool57}5859// Bus holds the connection's parts and the state that outlives one60// connection: the remembered subscriptions and the packet identifier61// counter. out is the current connection's write queue, or nil while62// disconnected, and the mutex guards both it and the fields around it.63type Bus struct {64	clientID  string65	will      *busWill66	onConnect func(*Bus)67	handler   busHandler68	dial      func(context.Context) (net.Conn, error)6970	mutex    sync.Mutex71	filters  map[string]struct{}72	out      chan []byte73	packetID uint1674}7576// newBus builds a client that dials the address over TCP. The address,77// the client identifier, the will, the connect callback, and the78// inbound handler are fixed for the client's life; the connection they79// drive is not.80func newBus(address, clientID string, will *busWill, onConnect func(*Bus), handler busHandler) *Bus {81	bus := &Bus{82		clientID:  clientID,83		will:      will,84		onConnect: onConnect,85		handler:   handler,86		filters:   map[string]struct{}{},87	}88	bus.dial = func(ctx context.Context) (net.Conn, error) {89		dialer := &net.Dialer{}90		return dialer.DialContext(ctx, "tcp", address)91	}92	return bus93}9495// Run holds the connection open until ctx ends. It dials, connects,96// and serves one session, then waits a backoff and dials again. A97// session that reached a CONNACK resets the backoff to its floor, so a98// connection that drops after an hour reconnects at once, while a99// broker that never answers is retried ever more slowly.100func (b *Bus) Run(ctx context.Context) {101	backoff := busMinBackoff102	for ctx.Err() == nil {103		connected := b.runSession(ctx)104		if ctx.Err() != nil {105			return106		}107		if connected {108			backoff = busMinBackoff109		}110		select {111		case <-ctx.Done():112			return113		case <-time.After(backoff):114		}115		if !connected {116			backoff *= 2117			if backoff > busMaxBackoff {118				backoff = busMaxBackoff119			}120		}121	}122}123124// runSession dials, completes the CONNECT handshake, and serves the125// connection until its reader or writer fails. It returns whether the126// handshake reached a CONNACK, which is what tells Run to reset the127// backoff.128func (b *Bus) runSession(parent context.Context) (connected bool) {129	conn, err := b.dial(parent)130	if err != nil {131		return false132	}133	defer conn.Close()134135	if _, err := conn.Write(encodeConnect(b.clientID, busKeepalive, b.will)); err != nil {136		return false137	}138	reader := bufio.NewReader(conn)139	first, body, err := readPacket(reader)140	if err != nil || first&0xF0 != mqttConnack {141		return false142	}143	if err := parseConnack(body); err != nil {144		return false145	}146147	ctx, cancel := context.WithCancel(parent)148	defer cancel()149	// The reader blocks in Read until the broker writes or the150	// connection closes. Closing the connection when the session ends151	// is what unblocks a reader waiting on a silent broker.152	defer context.AfterFunc(ctx, func() { conn.Close() })()153154	out := make(chan []byte, busQueueDepth)155	b.mutex.Lock()156	b.out = out157	b.mutex.Unlock()158	defer func() {159		b.mutex.Lock()160		b.out = nil161		b.mutex.Unlock()162	}()163164	var writing sync.WaitGroup165	writing.Add(1)166	go func() {167		defer writing.Done()168		// A write failure ends the session, so the reader stops too.169		defer cancel()170		b.writeLoop(ctx, conn, out)171	}()172173	// The remembered subscriptions go out first, so the broker is174	// delivering again before onConnect re-publishes any retained175	// state.176	b.resubscribe(out)177	if b.onConnect != nil {178		b.onConnect(b)179	}180181	b.readLoop(reader)182	cancel()183	writing.Wait()184	return true185}186187// writeLoop is the one goroutine that writes the connection. It sends188// each queued frame and, when no frame has gone out for half the189// keepalive, sends a PINGREQ so the broker hears from the client190// before the keepalive elapses.191func (b *Bus) writeLoop(ctx context.Context, conn net.Conn, out <-chan []byte) {192	idle := time.Duration(busKeepalive) * time.Second / 2193	ticker := time.NewTicker(idle)194	defer ticker.Stop()195	last := time.Now()196	for {197		select {198		case <-ctx.Done():199			return200		case frame := <-out:201			if _, err := conn.Write(frame); err != nil {202				return203			}204			last = time.Now()205		case <-ticker.C:206			if time.Since(last) >= idle {207				if _, err := conn.Write(encodePingreq()); err != nil {208					return209				}210				last = time.Now()211			}212		}213	}214}215216// readLoop reads whole packets and delivers each inbound PUBLISH to the217// handler. Any read error ends the loop and the session.218//219// A PINGRESP is read and dropped, because it only proves the broker is220// alive, and a successful read already shows that. A SUBACK is checked: a221// refused subscription is a reader that receives nothing for the life of222// the session, and the line in the pod log is the only sign of it.223func (b *Bus) readLoop(reader *bufio.Reader) {224	for {225		first, body, err := readPacket(reader)226		if err != nil {227			return228		}229		switch first & 0xF0 {230		case mqttPublish:231			topic, payload, ok := parsePublish(body)232			if ok && b.handler != nil {233				b.handler(topic, payload)234			}235		case mqttSuback:236			if err := parseSuback(body); err != nil {237				fmt.Fprintf(os.Stderr, "%v\n", err)238			}239		}240	}241}242243// Publish enqueues a QoS 0 PUBLISH from any goroutine. While the client244// is disconnected the queue does not exist and the publish is dropped,245// which is correct at QoS 0: a caller re-publishes its retained state246// from onConnect, so the broker holds the current value again within247// the reconnect.248func (b *Bus) Publish(topic string, payload []byte, retained bool) {249	b.mutex.Lock()250	out := b.out251	b.mutex.Unlock()252	if out == nil {253		return254	}255	select {256	case out <- encodePublish(topic, payload, retained):257	default:258	}259}260261// Subscribe remembers the filter and sends it if the client is262// connected. The remembered set is what runSession re-sends on every263// reconnect, so a subscription outlives the connection it was made on.264func (b *Bus) Subscribe(filter string) {265	b.mutex.Lock()266	b.filters[filter] = struct{}{}267	out := b.out268	frame := encodeSubscribe(b.nextPacketID(), filter)269	b.mutex.Unlock()270	if out == nil {271		return272	}273	select {274	case out <- frame:275	default:276	}277}278279// resubscribe sends every remembered filter on a fresh connection. The280// filters go out in sorted order so the frames a reconnect writes are281// the same every time, which keeps a test deterministic.282func (b *Bus) resubscribe(out chan<- []byte) {283	b.mutex.Lock()284	filters := make([]string, 0, len(b.filters))285	for filter := range b.filters {286		filters = append(filters, filter)287	}288	frames := make([][]byte, 0, len(filters))289	sort.Strings(filters)290	for _, filter := range filters {291		frames = append(frames, encodeSubscribe(b.nextPacketID(), filter))292	}293	b.mutex.Unlock()294	for _, frame := range frames {295		select {296		case out <- frame:297		default:298		}299	}300}301302// nextPacketID hands out the identifier a SUBSCRIBE carries and its303// SUBACK echoes. It skips zero, which the protocol reserves. The caller304// holds the mutex.305func (b *Bus) nextPacketID() uint16 {306	b.packetID++307	if b.packetID == 0 {308		b.packetID = 1309	}310	return b.packetID311}

1package main23// catalog.go is the write side of the catalog: a client that posts rows to its4// own sidecar's Corrosion agent over /v1/transactions. Reads come from the5// SQLite file directly and never through this client. Nothing writes the file6// outside the agent, because a write outside it corrupts the CRDT clocks, so7// every write is a batch of statements posted here.89import (10	"bytes"11	"context"12	"encoding/json"13	"fmt"14	"io"15	"net/http"16)1718// The transactions endpoint every write posts to.19const transactionsPath = "/v1/transactions"2021// The batch ceiling the proof of concept measured: 500 statements per request22// seeded 5000 titles in under half a second.23const maxBatch = 5002425// Catalog writes to one Corrosion agent. base is the agent's API address, bound26// to localhost in the pod.27type Catalog struct {28	base string29	http *http.Client30}3132// NewCatalog builds a client from the agent's base address and an HTTP client.33// A test hands in an httptest server's base and its client.34func NewCatalog(base string, httpClient *http.Client) *Catalog {35	return &Catalog{base: base, http: httpClient}36}3738// statement is one SQL statement and its bound parameters. Every value is a39// parameter and never concatenated into the SQL, so a title with a quote in it40// is data and not syntax.41type statement struct {42	sql    string43	params []any44}4546// transactionResponse is the agent's answer: one result per statement, each47// with the rows it changed or the error that stopped it.48type transactionResponse struct {49	Results []transactionResult `json:"results"`50}5152type transactionResult struct {53	RowsAffected int    `json:"rows_affected"`54	Error        string `json:"error"`55}5657// apply chunks the statements into requests of at most maxBatch and posts each.58// It returns the rows applied so far and stops at the first failure, so a caller59// sees both what landed and what broke.60func (c *Catalog) apply(ctx context.Context, statements []statement) (int, error) {61	applied := 062	for start := 0; start < len(statements); start += maxBatch {63		end := min(start+maxBatch, len(statements))64		n, err := c.post(ctx, statements[start:end])65		applied += n66		if err != nil {67			return applied, err68		}69	}70	return applied, nil71}7273// post sends one batch as a JSON array of [sql, [params...]] entries and reads74// the count applied. A non-2xx status or a per-statement error is a failure the75// caller sees.76func (c *Catalog) post(ctx context.Context, statements []statement) (int, error) {77	body := make([]any, len(statements))78	for i, s := range statements {79		params := s.params80		if params == nil {81			params = []any{}82		}83		body[i] = []any{s.sql, params}84	}85	// The payload is strings, numbers, and slices of them, so it always86	// marshals, and there is no failure here for a caller to answer.87	payload, _ := json.Marshal(body)8889	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.base+transactionsPath, bytes.NewReader(payload))90	if err != nil {91		return 0, err92	}93	req.Header.Set("Content-Type", "application/json")94	req.Header.Set("Accept", "application/json")9596	resp, err := c.http.Do(req)97	if err != nil {98		return 0, err99	}100	defer drain(resp.Body)101102	if resp.StatusCode < 200 || resp.StatusCode > 299 {103		message, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))104		return 0, fmt.Errorf("catalog transaction: %s: %s", resp.Status, message)105	}106107	var result transactionResponse108	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {109		return 0, fmt.Errorf("catalog transaction: decoding response: %w", err)110	}111	// One result per statement is the contract, so a short answer is a112	// failure and never a batch that applied.113	if len(result.Results) != len(statements) {114		return 0, fmt.Errorf("catalog transaction: %d results for %d statements",115			len(result.Results), len(statements))116	}117	applied := 0118	for _, r := range result.Results {119		if r.Error != "" {120			return applied, fmt.Errorf("catalog transaction: %s", r.Error)121		}122		applied += r.RowsAffected123	}124	return applied, nil125}126127// plainItemUpsert is the upsert for an item table that carries only the shared128// header and the slug, which series and sets do. Movies and episodes each129// carry a column beyond the header and have an upsert of their own. table is130// a constant this package names and never input, so naming it in the SQL text131// carries no injection.132//133// The conflict target is the whole primary key, (library, id), and the134// update names no primary-key column, because cr-sqlite reads a change135// to a key column as a delete and a create.136func plainItemUpsert(table string, params []any) statement {137	return statement{138		sql: `INSERT INTO ` + table + ` (library, id, kind, path, title, sort_key, released, added, art, duration, body, slug) ` +139			`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ` +140			`ON CONFLICT (library, id) DO UPDATE SET ` +141			`kind = excluded.kind, path = excluded.path, title = excluded.title, ` +142			`sort_key = excluded.sort_key, released = excluded.released, added = excluded.added, art = excluded.art, ` +143			`duration = excluded.duration, body = excluded.body, slug = excluded.slug`,144		params: params,145	}146}147148// itemParams marshals a body and lays the header out in column order.149//150// The library is the first parameter of every statement this file151// writes, because it leads every key.152func itemParams(library, id, kind, path, title, sortKey, released string, added int64, art string, duration int64, body any, slug string) []any {153	payload, _ := json.Marshal(body)154	return []any{library, id, kind, path, title, sortKey, released, added, art, duration, string(payload), slug}155}156157// The arts column: the list as JSON, and an empty list for an item with none,158// so a reader always parses a list.159func artsParam(arts []string) string {160	if len(arts) == 0 {161		return "[]"162	}163	payload, _ := json.Marshal(arts)164	return string(payload)165}166167// UpsertMovies writes movie rows in place, so a re-walk updates a title rather168// than dropping and recreating it. The movies table adds set_id after the169// header, so this upsert names its own columns.170func (c *Catalog) UpsertMovies(ctx context.Context, rows []movieRow) (int, error) {171	statements := make([]statement, len(rows))172	for i, row := range rows {173		params := itemParams(row.Library, row.Id, row.Kind, row.Path, row.Title, row.SortKey, row.Released, row.Added, row.Art, row.Duration, row.Body, row.Slug)174		statements[i] = statement{175			sql: `INSERT INTO movies (library, id, kind, path, title, sort_key, released, added, art, duration, body, slug, set_id, nfo_facts, arts) ` +176				`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ` +177				`ON CONFLICT (library, id) DO UPDATE SET ` +178				`kind = excluded.kind, path = excluded.path, title = excluded.title, ` +179				`sort_key = excluded.sort_key, released = excluded.released, added = excluded.added, art = excluded.art, ` +180				`duration = excluded.duration, body = excluded.body, slug = excluded.slug, ` +181				`set_id = excluded.set_id, nfo_facts = excluded.nfo_facts, arts = excluded.arts`,182			params: append(params, row.SetID, row.NFOFacts, artsParam(row.Arts)),183		}184	}185	return c.apply(ctx, statements)186}187188// UpsertSets writes the derived set rows in place, so a walk that reads a new189// member updates the set rather than dropping and recreating it.190func (c *Catalog) UpsertSets(ctx context.Context, rows []setRow) (int, error) {191	statements := make([]statement, len(rows))192	for i, row := range rows {193		params := itemParams(row.Library, row.Id, row.Kind, row.Path, row.Title, row.SortKey, row.Released, row.Added, row.Art, row.Duration, row.Body, row.Slug)194		statements[i] = plainItemUpsert("sets", params)195	}196	return c.apply(ctx, statements)197}198199// UpsertSeries writes series rows in place. The series table adds nfo_facts200// after the header, so this upsert names its own columns, as the movies one201// does.202func (c *Catalog) UpsertSeries(ctx context.Context, rows []seriesRow) (int, error) {203	statements := make([]statement, len(rows))204	for i, row := range rows {205		params := itemParams(row.Library, row.Id, row.Kind, row.Path, row.Title, row.SortKey, row.Released, row.Added, row.Art, row.Duration, row.Body, row.Slug)206		statements[i] = statement{207			sql: `INSERT INTO series (library, id, kind, path, title, sort_key, released, added, art, duration, body, slug, nfo_facts, arts) ` +208				`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ` +209				`ON CONFLICT (library, id) DO UPDATE SET ` +210				`kind = excluded.kind, path = excluded.path, title = excluded.title, ` +211				`sort_key = excluded.sort_key, released = excluded.released, added = excluded.added, art = excluded.art, ` +212				`duration = excluded.duration, body = excluded.body, slug = excluded.slug, ` +213				`nfo_facts = excluded.nfo_facts, arts = excluded.arts`,214			params: append(params, row.NFOFacts, artsParam(row.Arts)),215		}216	}217	return c.apply(ctx, statements)218}219220// UpsertEpisodes writes episode rows in place, with the three columns that place221// an episode under its series.222func (c *Catalog) UpsertEpisodes(ctx context.Context, rows []episodeRow) (int, error) {223	statements := make([]statement, len(rows))224	for i, row := range rows {225		payload, _ := json.Marshal(row.Body)226		statements[i] = statement{227			sql: `INSERT INTO episodes (library, id, kind, path, title, sort_key, released, added, art, duration, body, slug, series, season, episode, arts) ` +228				`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ` +229				`ON CONFLICT (library, id) DO UPDATE SET ` +230				`kind = excluded.kind, path = excluded.path, title = excluded.title, ` +231				`sort_key = excluded.sort_key, released = excluded.released, added = excluded.added, art = excluded.art, ` +232				`duration = excluded.duration, body = excluded.body, slug = excluded.slug, ` +233				`series = excluded.series, season = excluded.season, episode = excluded.episode, arts = excluded.arts`,234			params: []any{row.Library, row.Id, row.Kind, row.Path, row.Title, row.SortKey, row.Released, row.Added, row.Art, row.Duration, string(payload), row.Slug, row.Series, row.Season, row.Episode, artsParam(row.Arts)},235		}236	}237	return c.apply(ctx, statements)238}239240// UpsertFiles writes file rows in place. present is carried as 1 or 0, because241// the column is an integer.242func (c *Catalog) UpsertFiles(ctx context.Context, rows []fileRow) (int, error) {243	statements := make([]statement, len(rows))244	for i, row := range rows {245		present := 0246		if row.Present {247			present = 1248		}249		statements[i] = statement{250			sql: `INSERT INTO files (library, path, container, video_codec, audio_codec, width, height, size_bytes, duration_ms, trickplay, present, type, role, language, modified, arrived) ` +251				`VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ` +252				`ON CONFLICT (library, path) DO UPDATE SET ` +253				`container = excluded.container, video_codec = excluded.video_codec, ` +254				`audio_codec = excluded.audio_codec, width = excluded.width, height = excluded.height, ` +255				`size_bytes = excluded.size_bytes, duration_ms = excluded.duration_ms, trickplay = excluded.trickplay, present = excluded.present, ` +256				`type = excluded.type, role = excluded.role, language = excluded.language, modified = excluded.modified, arrived = excluded.arrived`,257			params: []any{row.Library, row.Path, row.Container, row.VideoCodec, row.AudioCodec, row.Width, row.Height, row.SizeBytes, row.DurationMs, row.Trickplay, present, row.Type, row.Role, row.Language, row.Modified, row.Arrived},258		}259	}260	return c.apply(ctx, statements)261}262263// UpsertFileItems writes the many-to-many link, one row per (file, item) pair a264// file carries.265//266// Every column of file_items is a primary-key column, so the row267// carries nothing to update and the upsert does nothing on a conflict.268// A repeat write changes no row and broadcasts nothing.269func (c *Catalog) UpsertFileItems(ctx context.Context, rows []fileRow) (int, error) {270	var statements []statement271	for _, row := range rows {272		for _, item := range row.Items {273			statements = append(statements, statement{274				sql:    `INSERT INTO file_items (library, path, item) VALUES (?, ?, ?) ON CONFLICT (library, path, item) DO NOTHING`,275				params: []any{row.Library, row.Path, item},276			})277		}278	}279	return c.apply(ctx, statements)280}281282// UpsertAliases writes alias rows in place, so every provider id and the folder283// key resolve to the item.284func (c *Catalog) UpsertAliases(ctx context.Context, rows []aliasRow) (int, error) {285	statements := make([]statement, len(rows))286	for i, row := range rows {287		statements[i] = statement{288			sql:    `INSERT INTO aliases (library, alias, item, source) VALUES (?, ?, ?, ?) ON CONFLICT (library, alias) DO UPDATE SET item = excluded.item, source = excluded.source`,289			params: []any{row.Library, row.Alias, row.Item, row.Source},290		}291	}292	return c.apply(ctx, statements)293}294295// deleteByKey builds one delete per key against a two-column primary296// key, the library and the table's own key, so a delete reaches one297// library's row and never another library's row of the same name.298// table and column are constants this package names and never input.299func deleteByKey(table, column, library string, keys []string) []statement {300	statements := make([]statement, len(keys))301	for i, key := range keys {302		statements[i] = statement{303			sql:    `DELETE FROM ` + table + ` WHERE library = ? AND ` + column + ` = ?`,304			params: []any{library, key},305		}306	}307	return statements308}309310// DeleteMovies removes movie rows whose titles left the volume.311func (c *Catalog) DeleteMovies(ctx context.Context, library string, ids []string) (int, error) {312	return c.apply(ctx, deleteByKey("movies", "id", library, ids))313}314315// DeleteSets removes set rows whose last member left the library.316func (c *Catalog) DeleteSets(ctx context.Context, library string, ids []string) (int, error) {317	return c.apply(ctx, deleteByKey("sets", "id", library, ids))318}319320// DeleteSeries removes series rows whose titles left the volume.321func (c *Catalog) DeleteSeries(ctx context.Context, library string, ids []string) (int, error) {322	return c.apply(ctx, deleteByKey("series", "id", library, ids))323}324325// DeleteEpisodes removes episode rows whose files left the volume.326func (c *Catalog) DeleteEpisodes(ctx context.Context, library string, ids []string) (int, error) {327	return c.apply(ctx, deleteByKey("episodes", "id", library, ids))328}329330// DeleteFiles removes file rows whose files left the volume.331func (c *Catalog) DeleteFiles(ctx context.Context, library string, paths []string) (int, error) {332	return c.apply(ctx, deleteByKey("files", "path", library, paths))333}334335// DeleteAliases removes alias rows a re-walk no longer produces.336func (c *Catalog) DeleteAliases(ctx context.Context, library string, aliases []string) (int, error) {337	return c.apply(ctx, deleteByKey("aliases", "alias", library, aliases))338}339340// The replicated tables of the schema: the set the read of every341// library the catalog holds covers. A table added to the schema is one342// entry here. The runs table is one of them, so a library that has lost343// every item row but whose last Job wrote a run is still a library the344// reporter reports on.345var catalogTables = []string{"aliases", "movies", "sets", "series", "episodes", "file_items", "files", "runs", "attempts",346	"contributors", "contributor_aliases", "credits", "genres",347	"franchises", "franchise_members", "franchise_runs"}348349// DeleteFileItems names all three columns of the link row, because all350// three are the primary key. A delete by fewer would take every other351// link the library, the file, or the item holds with it.352func (c *Catalog) DeleteFileItems(ctx context.Context, library string, links []fileItemKey) (int, error) {353	statements := make([]statement, len(links))354	for i, link := range links {355		statements[i] = statement{356			sql:    `DELETE FROM file_items WHERE library = ? AND path = ? AND item = ?`,357			params: []any{library, link.Path, link.Item},358		}359	}360	return c.apply(ctx, statements)361}

1package main23// The durable catalog volumes. There is one per Library, which4// its worker Jobs mount in turn, and one per namespace, which the5// catalog pod holds. Both are ReadWriteOnce, because one agent writes6// one SQLite database, and both are sized from the namespace Catalog,7// because every agent holds the whole namespace's catalog.89import (10	"context"11	"errors"12)1314// scannerCatalogClaimName is the durable catalog volume one Library's15// worker Jobs mount. It is derived from the Library name, so every pass16// names the same claim and the operator keeps no record of it. Its17// ReadWriteOnce is what serializes one library's Jobs.18//19// The claim must hold a database whose schema matches this release.20// Corrosion refuses to change the primary key of a database it already21// holds, and an old database started against a new schema starts22// quietly stale: it logs the refusal, serves the old tables, and fails23// every write of the new shape, one request at a time. The catalog is24// derived, so the cure is cheap: delete the claims when a release25// changes a primary key, and the next pass provisions fresh ones that26// one full walk refills.27func scannerCatalogClaimName(library string) string {28	return library + "-catalog"29}3031// buildCatalogClaim writes the catalog claim one Library's workers take.32// It is ReadWriteOnce, because one agent writes one SQLite database. It is33// sized from the namespace Catalog, because each agent holds the whole34// namespace's catalog. It is owned by the Library, so it survives a pod roll35// and is collected with the Library. An empty StorageClassName is omitted,36// so the cluster's default StorageClass binds it.37func buildCatalogClaim(library *Library, catalog *NamespaceCatalog) *PersistentVolumeClaim {38	return &PersistentVolumeClaim{39		APIVersion: claimAPIVersion,40		Kind:       "PersistentVolumeClaim",41		Metadata: ObjectMeta{42			Name:            scannerCatalogClaimName(library.Metadata.Name),43			Namespace:       library.Metadata.Namespace,44			Labels:          libraryLabels(library.Metadata.Name),45			OwnerReferences: []OwnerReference{libraryOwner(library)},46		},47		Spec: PersistentVolumeClaimSpec{48			AccessModes: []string{accessModeReadWriteOnce},49			Resources: VolumeResourceRequirements{50				Requests: map[string]string{"storage": catalogStorageSize(catalog)},51			},52			StorageClassName: catalog.Spec.Storage.StorageClassName,53		},54	}55}5657// standCatalogClaim creates the catalog claim when there is none and leaves58// an existing one alone. A PersistentVolumeClaim's spec is immutable once it59// binds, so the operator provisions the claim rather than reconciling it. A60// size a later Catalog grows to reaches a new claim, not this one. A conflict61// on the create means another writer got there first, which is success.62func (o *operator) standCatalogClaim(ctx context.Context, library *Library, catalog *NamespaceCatalog) error {63	namespace := library.Metadata.Namespace64	name := scannerCatalogClaimName(library.Metadata.Name)6566	_, err := GetPersistentVolumeClaim(ctx, o.client, namespace, name)67	if err == nil {68		return nil69	}70	if !errors.Is(err, ErrNotFound) {71		return err72	}7374	_, err = CreatePersistentVolumeClaim(ctx, o.client, buildCatalogClaim(library, catalog))75	if errors.Is(err, ErrConflict) {76		return nil77	}78	return err79}8081// The durable catalog volume the namespace's catalog pod holds,82// named from the Catalog, unless the Catalog names a claim of its own.83func catalogPodClaimName(catalog string) string {84	return catalog + "-catalog"85}8687// The claim the catalog pod mounts: the one the Catalog names,88// or the one the operator provisions when it names none.89func catalogClaimFor(catalog *NamespaceCatalog) string {90	if catalog.Spec.Storage.ClaimName != "" {91		return catalog.Spec.Storage.ClaimName92	}93	return catalogPodClaimName(catalog.Metadata.Name)94}9596// The catalog pod's own claim, owned by the Catalog, so the97// garbage collector takes it with the Catalog and the standing catalog98// survives every roll of the pod.99func buildCatalogPodClaim(catalog *NamespaceCatalog) *PersistentVolumeClaim {100	return &PersistentVolumeClaim{101		APIVersion: claimAPIVersion,102		Kind:       "PersistentVolumeClaim",103		Metadata: ObjectMeta{104			Name:            catalogPodClaimName(catalog.Metadata.Name),105			Namespace:       catalog.Metadata.Namespace,106			Labels:          catalogPodLabels(),107			OwnerReferences: []OwnerReference{catalogObjectOwner(catalog)},108		},109		Spec: PersistentVolumeClaimSpec{110			AccessModes: []string{accessModeReadWriteOnce},111			Resources: VolumeResourceRequirements{112				Requests: map[string]string{"storage": catalogStorageSize(catalog)},113			},114			StorageClassName: catalog.Spec.Storage.StorageClassName,115		},116	}117}

1package main23// The catalog pod is what a Catalog becomes at run time: one4// standing pod per namespace, owned by the Catalog, holding the5// namespace's durable catalog on a claim and reporting what it holds6// over the bus. It is the standing member of the gossip cluster, and7// every worker Job joins that cluster for the length of its run. It8// answers on no port: the agent's API is loopback only, and the9// reporter reads it from inside the pod.1011import (12	"context"13	"errors"14)1516// The pod one Catalog becomes, named from the Catalog, so every17// pass names the same pod and the operator keeps no record of it.18func catalogPodName(catalog string) string {19	return catalog + "-catalog"20}2122// The label pair the catalog pod carries: the name label that23// tells it from a Job's pod and a screen pod, and the member label the24// namespace's EndpointSlice is written over.25func catalogPodLabels() map[string]string {26	return withMemberLabel(map[string]string{scannerLabelKey: catalogLabelValue})27}2829// The pod the Catalog stands. It is a function of the Catalog30// and the operator's own settings alone, so two passes over an31// unchanged Catalog build the same pod, which is what makes the32// template hash mean anything.33func buildCatalogPod(catalog *NamespaceCatalog, scannerImage, corrosionImage, busAddress, topicBase string) *Pod {34	grace := int64(scannerGracePeriod)35	// The reporter holds no Kubernetes credential; it publishes over36	// the bus, and the operator alone writes a status.37	noToken := false38	return &Pod{39		APIVersion: podAPIVersion,40		Kind:       "Pod",41		Metadata: ObjectMeta{42			Name:            catalogPodName(catalog.Metadata.Name),43			Namespace:       catalog.Metadata.Namespace,44			Labels:          catalogPodLabels(),45			OwnerReferences: []OwnerReference{catalogObjectOwner(catalog)},46		},47		Spec: PodSpec{48			// The catalog pod is a standing service and not a run to49			// completion, so the kubelet restarts a container that50			// exits rather than letting the pod end.51			RestartPolicy:                 "Always",52			TerminationGracePeriodSeconds: &grace,53			AutomountServiceAccountToken:  &noToken,54			InitContainers: []Container{55				catalogSidecar(corrosionImage),56			},57			Containers: []Container{58				reporterSidecar(catalog, scannerImage, busAddress, topicBase),59			},60			Volumes: []Volume{61				{Name: catalogVolumeName, PersistentVolumeClaim: &PersistentVolumeClaimVolumeSource{62					ClaimName: catalogClaimFor(catalog),63				}},64			},65		},66	}67}6869// The container that reads the loopback catalog API and70// publishes each library's report over the bus. It runs this operator's71// own image in its report role, and it learns the namespace it reports72// on from its environment alone, because it holds no API credential.73func reporterSidecar(catalog *NamespaceCatalog, image, busAddress, topicBase string) Container {74	return Container{75		Name:    reporterContainer,76		Image:   image,77		Command: []string{"/library-operator", reportMode},78		Env: []EnvVar{79			{Name: libraryNamespaceVariable, Value: catalog.Metadata.Namespace},80			{Name: busAddressVariable, Value: busAddress},81			{Name: topicBaseVariable, Value: topicBase},82			{Name: catalogAPIVariable, Value: defaultCatalogAPI},83		},84		Resources: ResourceRequirements{85			Requests: map[string]string{"cpu": scannerCPURequest, "memory": scannerMemoryRequest},86			Limits:   map[string]string{"memory": scannerMemoryLimit},87		},88		SecurityContext: unprivileged(),89	}90}9192// The pod that stands for one Catalog after this pass, on the93// same terms as every other pod this operator stands: the live pod when94// it matches the template, the created pod when there was none, and nil95// when this pass deleted a stale one.96func (o *operator) standCatalogPod(ctx context.Context, catalog *NamespaceCatalog) (*Pod, error) {97	if err := o.standCatalogPodClaim(ctx, catalog); err != nil {98		return nil, err99	}100	desired := buildCatalogPod(catalog, o.scannerImage, o.corrosionImage, o.busAddress, o.topicBase)101	return o.standPod(ctx, desired)102}103104// The pod that holds the namespace's catalog, out of the pods105// the pass listed, or nil when it does not stand yet.106func catalogPodOf(catalog *NamespaceCatalog, pods []Pod) *Pod {107	if catalog == nil {108		return nil109	}110	name := catalogPodName(catalog.Metadata.Name)111	for index := range pods {112		pod := &pods[index]113		if pod.Metadata.Namespace == catalog.Metadata.Namespace && pod.Metadata.Name == name {114			return pod115		}116	}117	return nil118}119120// The reason and message the Catalog's Ready condition carries121// while its pod is not up, so a person reads one object to find what122// the namespace's catalog waits on.123func catalogPodBlocker(pod *Pod) (string, string) {124	switch {125	case pod == nil:126		return catalogReasonPodPending, "there is no catalog pod yet"127	case pod.Status.Phase == podFailed:128		return catalogReasonPodFailed, podFailureMessage(pod)129	case pod.Status.Phase != podRunning || !everyContainerReady(pod):130		return catalogReasonPodPending, podPendingMessage(pod)131	}132	return "", ""133}134135// An absent claim is created and an existing one is left alone,136// the rule standCatalogClaim follows, because a claim's spec is137// immutable once it binds. A Catalog that names a claim of its own138// creates none: the claim is the person's, and the operator mounts it.139func (o *operator) standCatalogPodClaim(ctx context.Context, catalog *NamespaceCatalog) error {140	if catalog.Spec.Storage.ClaimName != "" {141		return nil142	}143	namespace, name := catalog.Metadata.Namespace, catalogPodClaimName(catalog.Metadata.Name)144145	_, err := GetPersistentVolumeClaim(ctx, o.client, namespace, name)146	if err == nil {147		return nil148	}149	if !errors.Is(err, ErrNotFound) {150		return err151	}152	_, err = CreatePersistentVolumeClaim(ctx, o.client, buildCatalogPodClaim(catalog))153	if errors.Is(err, ErrConflict) {154		return nil155	}156	return err157}