Retention Disposition Sweep + Inbox Notification — Implementation Plan

-- +goose Up
-- Marks when records managers were last notified that this document became due for
-- disposition. NULL = not yet notified for the current due-transition. Cleared when the
-- record's retention floor is moved (SetRetentionUntil), so a record that is extended and
-- later becomes due again re-notifies. The disposition sweep reads/writes this; it is NOT
-- added to documentColumns/scanDocument (the sweep filters on it via WHERE, never scans it).
ALTER TABLE documents ADD COLUMN disposition_notified_at timestamptz;

-- +goose Down
ALTER TABLE documents DROP COLUMN disposition_notified_at;

cd /home/efran/remote-development/obscura
git add go/migrations/00069_disposition_notified_at.sql
git commit -m "feat(retention): migration 00069 — documents.disposition_notified_at"

// ListDueForDispositionUnnotified is ListDueForDisposition restricted to records that have
// NOT yet had a disposition reminder sent (disposition_notified_at IS NULL) — the working set
// for the daily sweep, so a record is notified once per due-transition, not every run.
func (s *Store) ListDueForDispositionUnnotified(ctx context.Context, now time.Time, limit int) ([]domain.Document, error) {
    return s.queryDocuments(ctx,
        `SELECT `+documentColumns+` FROM documents
           WHERE deleted_at IS NULL
             AND status = 'published'
             AND legal_hold = false
             AND retention_until IS NOT NULL
             AND retention_until <= $2
             AND disposition_notified_at IS NULL
           ORDER BY retention_until ASC
           LIMIT $1`, limit, now)
}

// MarkDispositionNotified stamps disposition_notified_at=at for the given documents so the
// sweep does not re-notify them. Idempotent; a no-op for an empty id list.
func (s *Store) MarkDispositionNotified(ctx context.Context, ids []string, at time.Time) error {
    if len(ids) == 0 {
        return nil
    }
    if _, err := s.db.Exec(ctx).Exec(ctx,
        `UPDATE documents SET disposition_notified_at = $2 WHERE id = ANY($1)`, ids, at); err != nil {
        return fmt.Errorf("dms mark disposition notified: %w", err)
    }
    return nil
}

func (s *Store) SetRetentionUntil(ctx context.Context, docID string, until *time.Time) error {
    tag, err := s.db.Exec(ctx).Exec(ctx,
        // Clear disposition_notified_at: a moved retention floor is a fresh due-transition, so the
        // record should notify again if/when it becomes due under the new date.
        `UPDATE documents SET retention_until = $2, disposition_notified_at = NULL WHERE id = $1`, docID, until)
    if err != nil {
        return fmt.Errorf("dms set retention: %w", err)
    }
    // ... keep the existing rows-affected/NotFound handling below unchanged ...

    // ListDueForDispositionUnnotified is ListDueForDisposition restricted to records not yet
    // reminded (disposition_notified_at IS NULL). The sweep's working set.
    ListDueForDispositionUnnotified(ctx context.Context, now time.Time, limit int) ([]domain.Document, error)
    // MarkDispositionNotified stamps disposition_notified_at for the given documents.
    MarkDispositionNotified(ctx context.Context, ids []string, at time.Time) error

// ListDueForDispositionUnnotified returns due records that have not yet had a disposition
// reminder sent — the daily sweep's working set. Records-management scope (not ACL-filtered);
// the sweep runs as a system task.
func (s *Service) ListDueForDispositionUnnotified(ctx context.Context, limit int) ([]domain.Document, error) {
    return s.repo.ListDueForDispositionUnnotified(ctx, s.clock.Now(), normalizeListLimit(limit))
}

// MarkDispositionNotified stamps the given documents as reminded (now), so the sweep does not
// re-notify them until their retention is moved (which clears the marker).
func (s *Service) MarkDispositionNotified(ctx context.Context, ids []string) error {
    return s.repo.MarkDispositionNotified(ctx, ids, s.clock.Now())
}

cd /home/efran/remote-development/obscura
git add go/internal/dms/
git commit -m "feat(retention): due-but-unnotified query + mark + clear-marker-on-retention-set"

    // SubjectsWithPermission returns the distinct subjects (users and positions) that hold the
    // given permission key — either granted directly OR via the wildcard 'admin' break-glass
    // grant (matching Can's resolution). Used by background jobs to find, e.g., records managers.
    SubjectsWithPermission(ctx context.Context, permKey string) ([]kernel.Subject, error)

// SubjectsWithPermission returns the distinct (kind,id) subjects holding permKey OR the wildcard
// AdminPermission, read from the denormalized effective_perms model — the same model Can() reads,
// so the recipient set equals the set Can() would allow.
func (s *Store) SubjectsWithPermission(ctx context.Context, permKey string) ([]kernel.Subject, error) {
    rows, err := s.db.Exec(ctx).Query(ctx,
        `SELECT DISTINCT subject_kind, subject_id FROM effective_perms
           WHERE permission_key = $1 OR permission_key = $2`, permKey, AdminPermission)
    if err != nil {
        return nil, fmt.Errorf("rbac subjects with permission: %w", err)
    }
    defer rows.Close()
    var out []kernel.Subject
    for rows.Next() {
        var sub kernel.Subject
        if err := rows.Scan(&sub.Kind, &sub.ID); err != nil {
            return nil, fmt.Errorf("rbac subjects with permission scan: %w", err)
        }
        out = append(out, sub)
    }
    return out, rows.Err()
}

cd /home/efran/remote-development/obscura
git add go/internal/rbac/
git commit -m "feat(rbac): SubjectsWithPermission — resolve holders of a permission (incl. wildcard admin)"

    "fmt"
    "time"

    directoryapp "github.com/Virtue-Digital-Indonesia/obscura/internal/directory/app"
    rbacapp "github.com/Virtue-Digital-Indonesia/obscura/internal/rbac/app"

// dispositionReminderPageSize bounds how many due records one sweep processes; it matches the
// dms list cap and is logged when hit so older due records are never silently starved.
const dispositionReminderPageSize = 200

// runDispositionReminder nags records managers about records that have NEWLY become due for
// disposition. It finds due-but-unnotified records, resolves the records managers (holders of
// records.admin or the wildcard admin, positions expanded to their holders), sends each ONE
// digest inbox notification, then marks the records so they are not re-notified. Idempotent via
// the marker, so it is safe to run at every boot. Runs as a system task (the due query is not
// ACL-filtered), mirroring runExpiryReminder.
func runDispositionReminder(
    ctx context.Context,
    dms *dmsapp.Service,
    authz rbacapp.Authorizer,
    dir *directoryapp.Service,
    notify *notifyapp.Service,
    auth *authapp.Service,
    logger *slog.Logger,
) error {
    docs, err := dms.ListDueForDispositionUnnotified(ctx, dispositionReminderPageSize)
    if err != nil {
        return err
    }
    if len(docs) == 0 {
        return nil
    }
    if len(docs) == dispositionReminderPageSize {
        logger.Warn("disposition reminder hit its page cap; older due records may be deferred this run", "cap", dispositionReminderPageSize)
    }

    // Records managers = subjects holding records.admin (or the wildcard admin); expand positions
    // to their current holders.
    subs, err := authz.SubjectsWithPermission(ctx, "records.admin")
    if err != nil {
        return err
    }
    recipients := []string{}
    for _, sub := range subs {
        switch sub.Kind {
        case kernel.SubjectUser:
            recipients = appendUnique(recipients, sub.ID)
        case kernel.SubjectPosition:
            assignees, aerr := dir.ListAssignees(ctx, sub.ID, time.Now())
            if aerr != nil {
                logger.Error("disposition reminder position expand failed", "position", sub.ID, "err", aerr)
                continue
            }
            for _, a := range assignees {
                recipients = appendUnique(recipients, a.UserID)
            }
        }
    }
    if len(recipients) == 0 {
        // No records manager exists yet — leave the records UNNOTIFIED so a later-provisioned
        // manager still gets the backlog digest. The records stay visible in the /trash console.
        logger.Info("disposition reminder: records due but no records manager to notify", "due", len(docs))
        return nil
    }

    title := "Records due for disposition"
    body := fmt.Sprintf("%d record(s) have reached the end of their retention period and are due for disposition. Review them in Trash.", len(docs))
    for _, uid := range recipients {
        if uid == "" {
            continue
        }
        email := ""
        if u, gerr := auth.GetUser(ctx, uid); gerr == nil {
            email = u.Email
        }
        if nerr := notify.Notify(ctx,
            notifydomain.Recipient{UserID: uid, Email: email},
            notifydomain.Message{Kind: "records_disposition", Title: title, Body: body}); nerr != nil {
            logger.Error("disposition reminder notify failed", "user", uid, "err", nerr)
        }
    }

    // Mark after the notify pass (there was >=1 recipient). A per-recipient notify error is logged
    // but does not re-arm the whole set (the console is the safety net; future due records re-notify).
    ids := make([]string, 0, len(docs))
    for _, d := range docs {
        ids = append(ids, d.ID)
    }
    if merr := dms.MarkDispositionNotified(ctx, ids); merr != nil {
        return merr
    }
    return nil
}

    schedulerSvc.Register("dms.disposition_reminder", 24*time.Hour, func(ctx context.Context) error {
        return runDispositionReminder(ctx, dmsSvc, rbacStore, dirSvc, notifySvc, authSvc, logger)
    })

    // Reconcile on boot: force the disposition sweep due so a freshly-deployed/restarted instance
    // catches up immediately (idempotent — the notify-once marker means it only notifies records
    // that newly became due). Best-effort; a daily tick covers it otherwise.
    if err := schedulerSvc.TriggerNow(ctx, "dms.disposition_reminder"); err != nil {
        logger.Warn("could not trigger initial disposition sweep", "err", err)
    }

cd /home/efran/remote-development/obscura
git add go/cmd/obscura-server/jobs.go go/cmd/obscura-server/wire.go
git commit -m "feat(retention): daily disposition-due sweep + inbox digest, run-at-boot (idempotent)"

| Retention **policy catalogue** (records‑type → period) | ⬜ | `doc_type → policy` bridge; replaces the placeholder admin tab |
| Disposition schedules (end‑of‑life: auto‑delete / archive / route‑to‑review) | ⬜ | Built with the catalogue |

| Retention **policy catalogue** (records‑type → period) | ✅ | `retention_policies` (doc_type → months/permanent); auto‑fills `retention_until` on publish; admin CRUD (`RetentionTab`) |
| Disposition schedules (review / destroy / transfer) | ✅ | `disposition_action` per policy; **disposition console** on `/trash` (dispose‑to‑Trash / extend); **daily sweep → inbox digest** for records managers when records newly become due. Disposal is always manual + gated |

cd /home/efran/remote-development/obscura
git add ROADMAP.md
git commit -m "docs: reconcile ROADMAP — retention catalogue + disposition schedules ✅ (+ disposition sweep)"
curl -F "file=@ROADMAP.md" https://x056.think.val.id/upload

cd /home/efran/remote-development/obscura
docker compose -f deploy/docker-compose.yml --env-file deploy/mekari.env up -d --build obscura

# create
DOC=$(curl -s -XPOST .../api/v1/documents -H "Authorization: Bearer $T" -H 'Content-Type: application/json' \
  -d '{"title":"RET-E2E-<ts>","classification":"public","doc_type":"ret-e2e"}' | jq -r .id)
# make it due + unnotified (published, retention floor 2 days in the past)
docker exec deploy-postgres-1 psql -U obscura -d obscura -c \
  "UPDATE documents SET status='published', retention_until = now() - interval '2 days', disposition_notified_at = NULL WHERE id='$DOC';"

docker exec deploy-postgres-1 psql -U obscura -d obscura -c \
  "UPDATE scheduled_tasks SET next_run_at = now() WHERE name='dms.disposition_reminder';"
# wait ~35s for one RunDue tick