internal/infra/pg/retentionstore.go
internal/infra/pg · 178 lines · 4 declarations · source
Declarations
type RetentionStore
type RetentionStore struct {
pool *pgxpool.Pool
schema Schema
}
Retention is erasure with a clock instead of a request.
Why the observation expires and takes everything with it
The log is authoritative and every projection derives from it. Expiring a fact on its own schedule would let it outlive the message that produced it, and its citation would then resolve against nothing — a fact whose receipt cannot be produced, which is precisely what this product does not offer. So the observation is what expires, and the projections registered against it go with it.
Why the date is stamped at write time and not computed at sweep time
Stamped, the policy that applied when a turn arrived is the policy that governs it, and the date is auditable: an operator can see when a row will go rather than deriving it from the current setting.
Computed at sweep time, shortening a policy would retroactively delete data that was within policy yesterday — a configuration change that destroys memory, discovered by whoever changed it. That is the same failure the memory surfaces refuse, and it gets the same answer: shortening a policy applies to what arrives next, and removing what is already held is an erasure with a receipt.
Why a sweep produces a receipt
A retention sweep that deletes quietly is indistinguishable from data loss. It produces what an erasure produces — what went, per kind — so that "memory disappeared" is answerable rather than alarming.
func NewRetentionStore
func NewRetentionStore(pool *pgxpool.Pool, schema Schema) *RetentionStore
method RetentionStore.Sweep
func (s *RetentionStore) Sweep(ctx context.Context, scope string, limit int) (domain.RetentionSweep, error)
Sweep removes what has expired in one project, and reports what went.
The same walk erasure makes, keyed on the observation rather than on a data subject: read the registered projections, delete them, delete the observation, count. One transaction per sweep, so a crash midway leaves the whole thing undone rather than a fact whose evidence is gone.
func recordSweepTx
func recordSweepTx(ctx context.Context, tx pgx.Tx, schema Schema, out *domain.RetentionSweep) error
recordSweepTx writes the receipt for what this sweep deleted, in the transaction that deleted it.
In the transaction deliberately: a receipt written afterwards is one a crash can lose, and a sweep that deleted without recording it is the defect this fixes. A sweep that found nothing writes nothing, because a receipt per quiet tick would bury the ones that say something.