# Topic System Plan — Typed topics, state modifiers, and live held-state lookup

## Overview

A server-side pub/sub system that lets external services (workers, webhooks,
other actions) push updates to clients watching a specific entity. Built on top
of the existing slot/SSE pipeline — no new transport.

A **topic** (`Topic<TPayload>`) loads the entity from the database when
triggered. **State modifiers** (`StateModifier<TPayload, TState>`) receive the
payload + the watcher's current slot state + a live lookup of all the client's
held states, and decide whether to push an update (return new state), skip
(return `current`), or revoke (return `null`).

---

## Design

### `Topic<TPayload>` — loads the entity once per trigger

```vala
public abstract class Topic<TPayload> : Object {
    /** Load the payload for `topic_key` (e.g. "cat:42" → Cat from the database). */
    public abstract async TPayload load(string topic_key) throws GLib.Error;
}
```

Runs once per trigger (not per watcher). DI-constructed — can inject databases,
APIs, etc.

### `StateModifier<TPayload, TState>` — derives slot state per watcher

```vala
public abstract class StateModifier<TPayload, TState> : Object {
    /**
     * @param payload  The trigger payload (loaded once by Topic<T>).
     * @param current  The current state of the slot being modified, or null if
     *                 it has expired from the cache.
     * @param held     Live lookup of the client's currently-held states by
     *                 type_name (reads CURRENT cache values at trigger time).
     * @return New TState → push; `current` (same ref) → skip; null → clear + remove.
     */
    public abstract TState? derive(TPayload payload, TState? current, HeldStates held);
}
```

### `HeldStates` — live lookup of client state at trigger time

```vala
public class HeldStates : Object {
    /** Typed current state for a held type, or null if expired/dropped. */
    public T? get<T>(string type_name);
}
```

Internally holds a `type_name → slot_key` map (captured at watch time from
`request.held`). On each `get<T>`, reads the CURRENT state from
`state_service.get_slot(key)`. If the slot expired, returns null.

### Return-value semantics

| `derive` returns | Action |
|---|---|
| New `TState` instance | `state_service.update(slot_key, state)` → SSE push |
| `current` (same reference) | No push (unchanged / not relevant) |
| `null` | `state_service.clear_slot(slot_key)` → SSE clear + remove watcher |
| Throws exception | Same as `null`: clear slot + remove watcher |

---

## Registration

```vala
// Startup
statum.topic<CatTopic>("cat");
statum.topic_modifier<CatBreedDetailModifier, Cat, CatBreedDetail>("cat");
statum.topic_modifier<CatSummaryModifier, Cat, CatSummary>("cat");
```

`statum.topic<T>(prefix)` registers the topic handler for a key prefix (e.g.
"cat" matches "cat:42", "cat:99").

`statum.topic_modifier<M, TPayload, TState>(prefix)` registers a state modifier
for the prefix. Multiple modifiers can share a prefix (one per slot type).

## Entrypoint — watch with held context

```vala
public class CatDetailEntrypoint : StatumEntrypoint {
    private TopicRegistry topics = inject<TopicRegistry>();

    public override async DirectiveBuilder handle() throws GLib.Error {
        var cat = yield db.find_cat_by_breed(request.route("breed"));

        var detail_slot = state_service.new_slot(Scope.PAGE, to_detail_state(cat));
        var summary_slot = state_service.new_slot(Scope.PAGE, to_summary_state(cat));

        // Capture the full held context (type_name → slot_key) so modifiers
        // can query auth, related entities, etc. at trigger time.
        topics.watch(@"cat:$(cat.id)", detail_slot.id, typeof(CatBreedDetail).name(), request.held);
        topics.watch(@"cat:$(cat.id)", summary_slot.id, typeof(CatSummary).name(), request.held);

        return directives()
            .set(detail_slot)
            .set(summary_slot)
            .subscribe(detail_slot.id)
            .subscribe(summary_slot.id);
    }
}
```

## Trigger

```vala
// From any context (worker, webhook, admin action)
yield topics.trigger("cat:42");
```

## What the registry does on trigger

1. Resolve the topic handler by prefix ("cat" → `CatTopic`).
2. Check if there are any watchers for `topic_key`. If none, **skip the load
   entirely** — no database query, no DI construction. The trigger is a no-op.
3. `topic.load("cat:42")` → `Cat` (one DB query).
4. Find all modifiers for the prefix.
5. For each watcher:
   a. Find the modifier whose `state_type_name` matches the watcher's slot type.
   b. Read `current` from `state_service.get_slot(watcher.slot_key)`.
   c. Build `HeldStates` from `watcher.held_keys`.
   d. Call `modifier.derive(payload, current, held)`.
   e. If result is null OR the call throws → `clear_slot` + remove watcher.
   f. If result == `current` (reference equality) → skip.
   g. If result is a new instance → `update(slot_key, state)` → SSE push.

## Cleanup

| Event | Action |
|---|---|
| `derive` returns null | `clear_slot` + remove watcher |
| `derive` throws | Same as null: `clear_slot` + remove watcher |
| SSE disconnect (navigate away) | `update` push is no-op; periodic sweep removes stale watchers |
| Server restart | Watchers lost from memory; re-registered on next entrypoint load |
| Slot expired from cache | `trigger` catches `SLOT_NOT_FOUND` in `update` → removes watcher |
| Periodic sweep (every 5 min) | Checks `get_slot(key)` for each watcher; removes if null |

---

## Tasks

### V-1: `HeldStates` — `src/TopicSystem.vala` (or `src/HeldStates.vala`)

```vala
public class HeldStates : Object {
    private Dictionary<string, string> slot_keys;
    private StateService state_service;

    internal HeldStates(Dictionary<string, string> slot_keys, StateService state_service) { ... }

    public T? get<T>(string type_name) { ... }
    public Properties? get_properties(string type_name) { ... }
}
```

`get<T>` reads the current state from `state_service.get_slot`, maps to `T` via
`GObjectMapping.from_properties_typed<T>`.

### V-2: `Topic<TPayload>` base class — `src/TopicSystem.vala`

```vala
public abstract class Topic<TPayload> : Object {
    public abstract async TPayload load(string topic_key) throws GLib.Error;
}
```

### V-3: `StateModifier<TPayload, TState>` + non-generic bridge — `src/TopicSystem.vala`

The generic class stores the payload/state types. A non-generic internal
interface (`TopicStateModifier`) lets the registry store heterogeneous modifiers:

```vala
internal interface TopicStateModifier : Object {
    public abstract string state_type_name { get; }
    public abstract async ModifierOutcome apply(Object payload, string slot_key, HeldStates held) throws GLib.Error;
}

public enum ModifierOutcome { UPDATED, UNCHANGED, CLEARED }

public abstract class StateModifier<TPayload, TState> : Object, TopicStateModifier {
    protected StateService state_service = inject<StateService>();

    public string state_type_name { get { return typeof(TState).name(); } }

    public abstract TState? derive(TPayload payload, TState? current, HeldStates held);

    public async ModifierOutcome apply(Object payload, string slot_key, HeldStates held) throws GLib.Error {
        var slot = state_service.get_slot(slot_key);
        TState? current = null;
        if (slot != null && slot.current_state != null) {
            try {
                current = GObjectMapping.from_properties_typed<TState>(typeof(TState), slot.current_state.public_data);
            } catch { current = null; }
        }

        var result = derive((TPayload) payload, current, held);

        if (result == null) {
            yield state_service.clear_slot(slot_key);
            return ModifierOutcome.CLEARED;
        }
        if (result == current) {
            return ModifierOutcome.UNCHANGED;
        }

        var new_state = new State() {
            type_name = state_type_name,
            public_data = GObjectMapping.to_properties((Object) result),
            private_data = slot?.current_state?.private_data ?? new PropertyDictionary()
        };
        yield state_service.update(slot_key, new_state);
        return ModifierOutcome.UPDATED;
    }
}
```

### V-4: `TopicRegistry` (singleton) — `src/TopicSystem.vala`

```vala
public class TopicRegistry : Object {
    private StateService state_service = inject<StateService>();

    // topic_key → watchers
    private Dictionary<string, Series<Watcher>> topic_watchers;
    // topic prefix → Topic handler (stored as a factory, resolved via DI)
    private Dictionary<string, Type> topic_types;
    // topic prefix → modifiers (by state_type_name)
    private Dictionary<string, Dictionary<string, TopicStateModifier>> prefix_modifiers;

    // Watcher record
    private class Watcher : Object {
        public string slot_key;
        public string slot_type;
        public Dictionary<string, string> held_keys; // type_name → slot_key
    }

    public void watch(string topic_key, string slot_key, string slot_type,
                      ReadOnlyAssociative<string, HeldSlot> request_held) { ... }

    public async void trigger(string topic_key) throws GLib.Error {
        // Early exit: if nobody is watching, skip the load entirely (no DB query).
        var watchers = topic_watchers.get_or_default(topic_key);
        if (watchers == null || ((!)watchers).to_array().length == 0) return;

        var prefix = extract_prefix(topic_key); // "cat:42" → "cat"
        var topic = construct_topic(prefix);
        var payload = yield topic.load(topic_key);

        var to_remove = new Series<Watcher>();
        foreach (var watcher in (!)watchers) {
            var modifiers = prefix_modifiers.get_or_default(prefix);
            if (modifiers == null) continue;

            TopicStateModifier modifier;
            if (!((!)modifiers).try_get(watcher.slot_type, out modifier)) continue;

            var held = new HeldStates(watcher.held_keys, state_service);

            try {
                var outcome = yield modifier.apply(payload, watcher.slot_key, held);
                if (outcome == ModifierOutcome.CLEARED) {
                    to_remove.add(watcher);
                }
            } catch (GLib.Error e) {
                // Exception in derive → clear + remove (same as null return)
                try { yield state_service.clear_slot(watcher.slot_key); }
                catch {}
                to_remove.add(watcher);
            }
        }

        foreach (var w in to_remove) {
            ((!)watchers).remove(w);
        }
    }

    // Periodic cleanup: remove watchers whose slots are no longer cached.
    public void sweep() { ... }
}
```

### V-5: `StatumConfigurator` — register topics + modifiers

```vala
public void topic<TTopic>(string prefix) throws Error {
    container.register_transient<TTopic>();
    // Store typeof(TTopic) for the prefix
}

public void topic_modifier<TModifier, TPayload, TState>(string prefix) throws Error {
    container.register_transient<TModifier>();
    // Store typeof(TModifier) for the prefix + typeof(TState).name()
}
```

The registry resolves modifiers and topics from the container (DI-constructed).

### V-6: `StatumModule` — register `TopicRegistry` singleton + sweep timer

```vala
container.register_singleton<TopicRegistry>();
// Sweep timer runs every 5 minutes (background).
```

### V-7: Entrypoint/action access

Both `StatumEntrypoint` and `StatumAction` gain:
```vala
protected TopicRegistry topics = inject<TopicRegistry>();
```

### V-8: Example

Add a `CatTopic` + `CatBreedDetailModifier` to the example (or a simpler variant
using the existing announcement/counter models to demonstrate the flow without a
database).

### D-1: Documentation

- **`model.md`**: add a "Topic system" section.
- **`getting-started.md`**: add `Topic<T>`, `StateModifier<T,S>`, `TopicRegistry`
  to the Vala API reference.

---

## API sketch (end-to-end)

```vala
// Topic: loads the entity
public class CatTopic : Topic<Cat> {
    private Database db = inject<Database>();
    public override async Cat load(string topic_key) throws GLib.Error {
        return yield db.load_cat(int.parse(topic_key.split(":")[1]));
    }
}

// Modifier: derives CatBreedDetail from Cat + held context
public class CatBreedDetailModifier : StateModifier<Cat, CatBreedDetail> {
    public override CatBreedDetail? derive(Cat cat, CatBreedDetail? current, HeldStates held) {
        var auth = held.get<AuthPublic>("auth");
        if (auth == null) return null;                       // expired → clear
        if (current != null && current.breed != cat.breed)
            return current;                                  // different breed → skip
        return new CatBreedDetail() { ... };                 // push
    }
}

// Registration
statum.topic<CatTopic>("cat");
statum.topic_modifier<CatBreedDetailModifier, Cat, CatBreedDetail>("cat");

// Entrypoint
topics.watch(@"cat:$(id)", slot.id, typeof(CatBreedDetail).name(), request.held);

// Trigger (anywhere)
yield topics.trigger("cat:42");
```

## Server resources

- Each watcher: ~150 bytes (topic key + slot key + type name + held keys map).
- 10K concurrent watchers ≈ 1.5 MB.
- Trigger cost: one `load()` (DB) + per-watcher: one `get_slot` + one `derive` (in-memory) + optionally one `update` (sign + push).
- Sweep: every 5 minutes, iterates watchers, checks `get_slot`. O(N) where N = watcher count.

## Out of scope

- Per-watcher personalised payloads (different state for different watchers of the same topic). Use per-user topic keys instead.
- Cross-process pub/sub (Redis/NATS backend for multi-server). Each process has its own TopicRegistry.
- Pattern-based topic matching (wildcards). Prefix matching ("cat" matches "cat:42") is sufficient.