feat(quartz): emit StoreEvent.Delete from observable for filter / expired sweeps
Replaces ObservableEventStore.events: SharedFlow<Event> with
SharedFlow<StoreEvent>, where StoreEvent is:
Insert(event)
Delete(rule: DeleteRule)
DeleteRule.Filtered(filters) ← from delete(filter) / delete(filters)
DeleteRule.Expired(asOf) ← from deleteExpiredEvents()
ObservableEventStore now emits Insert for accepted events, and Delete
for every out-of-band removal that went through it. The Expired rule
carries the cutoff timestamp (TimeUtils.now() at call time) so the
projection's in-memory drop matches the store's on-disk drop without
clock skew between collectors.
EventStoreProjection no longer runs its own NIP-40 ticker. Expired
events linger in [items] until the application calls
deleteExpiredEvents() on the observable — at which point the
projection drops everything whose expiration is past `asOf`. The
ticker, expirationTickMs constructor arg, and sweepExpired helper are
all removed.
For DeleteRule.Filtered the projection iterates current slots and
drops any whose event matches any of the rule's filters via
Filter.match — same predicate the store uses.
Tests:
- nip40ExpirationDroppedByTicker → nip40ExpirationDroppedOnStoreSweep:
drives a deleteExpiredEvents() call instead of waiting on a ticker.
- New deleteByFilterRemovesMatchingSlots: confirms delete(filter) on
the observable removes exactly the matching slots from the projection
without touching others.
- 15/15 projection tests + all other store tests pass.
https://claude.ai/code/session_01Jny85MTu1ynKgFBgysfWu5
This commit is contained in:
+46
-13
@@ -349,12 +349,12 @@ class EventStoreProjectionTest {
|
||||
}
|
||||
|
||||
/**
|
||||
* NIP-40 per-projection ticker: a slot whose `expiration` lapses
|
||||
* after the projection has loaded should be dropped on the next
|
||||
* tick, even though the store hasn't run its sweep yet.
|
||||
* NIP-40 expiration drops slots only when the application calls
|
||||
* `deleteExpiredEvents()` on the observable store — projections
|
||||
* no longer run their own ticker.
|
||||
*/
|
||||
@Test
|
||||
fun nip40ExpirationDroppedByTicker() =
|
||||
fun nip40ExpirationDroppedOnStoreSweep() =
|
||||
runBlocking {
|
||||
val time = TimeUtils.now()
|
||||
val safe = signer.sign(TextNoteEvent.build("safe", createdAt = time) { expiration(time + 100) })
|
||||
@@ -362,26 +362,59 @@ class EventStoreProjectionTest {
|
||||
observable.insert(safe)
|
||||
observable.insert(short)
|
||||
|
||||
// Drive the ticker frequently so the test doesn't sit idle.
|
||||
val projection =
|
||||
EventStoreProjection<Event>(
|
||||
observable,
|
||||
listOf(Filter(kinds = listOf(TextNoteEvent.KIND))),
|
||||
relay = null,
|
||||
scope = scope,
|
||||
expirationTickMs = 100,
|
||||
observable.observe<Event>(
|
||||
Filter(kinds = listOf(TextNoteEvent.KIND)),
|
||||
store.relay,
|
||||
scope,
|
||||
)
|
||||
projection.ready.await()
|
||||
assertEquals(2, projection.items.value.size)
|
||||
|
||||
// Wait past the short expiration.
|
||||
// Let the short expiration lapse, then ask the store to
|
||||
// sweep — the projection drops the expired slot in
|
||||
// response to the resulting StoreEvent.Delete(Expired).
|
||||
delay(2000)
|
||||
observable.deleteExpiredEvents()
|
||||
|
||||
val after = projection.awaitItems(timeoutMs = 5_000) { it.size == 1 }
|
||||
val after = projection.awaitItems { it.size == 1 }
|
||||
assertEquals(safe.id, after[0].value.id)
|
||||
projection.close()
|
||||
}
|
||||
|
||||
/**
|
||||
* `delete(filter)` on the observable propagates to open
|
||||
* projections: they drop every slot matching the filter using
|
||||
* the same Filter.match logic the store would.
|
||||
*/
|
||||
@Test
|
||||
fun deleteByFilterRemovesMatchingSlots() =
|
||||
runBlocking {
|
||||
val a = signer.sign(TextNoteEvent.build("a", createdAt = 100))
|
||||
val b = signer.sign(TextNoteEvent.build("b", createdAt = 200))
|
||||
val foreign = otherSigner.sign(TextNoteEvent.build("foreign", createdAt = 150))
|
||||
observable.insert(a)
|
||||
observable.insert(b)
|
||||
observable.insert(foreign)
|
||||
|
||||
val projection =
|
||||
observable.observe<Event>(
|
||||
Filter(kinds = listOf(TextNoteEvent.KIND)),
|
||||
store.relay,
|
||||
scope,
|
||||
)
|
||||
projection.ready.await()
|
||||
assertEquals(3, projection.items.value.size)
|
||||
|
||||
// Drop everything authored by `signer` — should leave
|
||||
// only the foreign event.
|
||||
observable.delete(Filter(authors = listOf(signer.pubKey)))
|
||||
|
||||
val after = projection.awaitItems { it.size == 1 }
|
||||
assertEquals(foreign.id, after[0].value.id)
|
||||
projection.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun limitIsEnforcedOnInsertOverflow() =
|
||||
runBlocking {
|
||||
|
||||
Reference in New Issue
Block a user