What is the value now?

What did we believe on date X about facts effective on date Y?

Case C-100 was closed effective 2026-03-31.
The closure was entered into the case system on 2026-04-02.
The closure was corrected on 2026-04-10 because the effective date should have been 2026-03-30.
A report generated on 2026-04-05 used the old effective date.
A report generated on 2026-04-12 should use the corrected date.

valid time      = when the fact is true in the real/business world
transaction time = when the system recorded or believed the fact

recorded_at: 2026-04-02T10:00:00Z
valid_from:  2026-03-31T00:00:00Z
valid_to:    infinity
case_id:     C-100
status:      CLOSED

recorded_at: 2026-04-10T09:00:00Z
valid_from:  2026-03-30T00:00:00Z
valid_to:    infinity
case_id:     C-100
status:      CLOSED
correction_of: previous fact
reason:      EFFECTIVE_DATE_CORRECTION

business_key
valid_from
valid_to
recorded_from
recorded_to
payload
assertion_id
supersedes_assertion_id
reason
source_lineage

[from, to)

start inclusive, end exclusive

valid_from <= query_valid_time < valid_to
recorded_from <= query_recorded_time < recorded_to

Event emitted: 2026-04-02
Decision effective date: 2026-03-31

source recorded_at: 2026-04-02 10:00
pipeline processed_at: 2026-04-03 01:00

Never mutate a past assertion silently.
Add a new assertion that supersedes or corrects it.

update case_status
set effective_date = '2026-03-30'
where case_id = 'C-100';

insert into case_status_assertion (..., valid_from, recorded_from, supersedes_assertion_id, reason)
values (..., '2026-03-30', now(), 'old-assertion-id', 'EFFECTIVE_DATE_CORRECTION');

update case_status_assertion
set recorded_to = now()
where assertion_id = 'old-assertion-id';

create table case_status_bitemporal (
    assertion_id varchar primary key,
    case_id varchar not null,
    status varchar not null,

    valid_from timestamp not null,
    valid_to timestamp not null,

    recorded_from timestamp not null,
    recorded_to timestamp not null,

    source_event_id varchar not null,
    source_system varchar not null,
    source_commit_time timestamp null,

    supersedes_assertion_id varchar null,
    correction_reason varchar null,
    produced_by_run_id varchar not null,
    transform_version varchar not null,

    payload_hash varchar not null,
    created_at timestamp not null
);

9999-12-31T00:00:00Z

create index idx_case_status_valid
on case_status_bitemporal (case_id, valid_from, valid_to);

create index idx_case_status_recorded
on case_status_bitemporal (case_id, recorded_from, recorded_to);

create index idx_case_status_bitemporal_query
on case_status_bitemporal (case_id, valid_from, valid_to, recorded_from, recorded_to);

select *
from case_status_bitemporal
where case_id = 'C-100'
  and valid_from <= current_timestamp
  and current_timestamp < valid_to
  and recorded_to = timestamp '9999-12-31 00:00:00';

currently valid and currently accepted

select *
from case_status_bitemporal
where case_id = 'C-100'
  and valid_from <= timestamp '2026-03-31 12:00:00'
  and timestamp '2026-03-31 12:00:00' < valid_to
  and recorded_to = timestamp '9999-12-31 00:00:00';

select *
from case_status_bitemporal
where case_id = 'C-100'
  and valid_from <= timestamp '2026-03-31 12:00:00'
  and timestamp '2026-03-31 12:00:00' < valid_to
  and recorded_from <= timestamp '2026-04-05 00:00:00'
  and timestamp '2026-04-05 00:00:00' < recorded_to;

What did we believe on April 5 about the status valid on March 31?

{
  "eventId": "evt-correction-001",
  "eventType": "CaseStatusCorrected",
  "caseId": "C-100",
  "correctsEventId": "evt-status-777",
  "correctionReason": "EFFECTIVE_DATE_CORRECTION",
  "old": {
    "status": "CLOSED",
    "effectiveFrom": "2026-03-31T00:00:00Z"
  },
  "new": {
    "status": "CLOSED",
    "effectiveFrom": "2026-03-30T00:00:00Z"
  },
  "recordedAt": "2026-04-10T09:00:00Z",
  "sourceCommitTime": "2026-04-10T09:00:02Z",
  "causationId": "cmd-correct-status-001",
  "correlationId": "case-C-100-correction-20260410"
}

The ledger is the durable truth.
Projections are views derived from the ledger.

public record ValidTimeRange(Instant fromInclusive, Instant toExclusive) {
    public ValidTimeRange {
        if (!fromInclusive.isBefore(toExclusive)) {
            throw new IllegalArgumentException("valid time range must be non-empty");
        }
    }

    public boolean contains(Instant t) {
        return !t.isBefore(fromInclusive) && t.isBefore(toExclusive);
    }
}

public record RecordedTimeRange(Instant fromInclusive, Instant toExclusive) {
    public RecordedTimeRange {
        if (!fromInclusive.isBefore(toExclusive)) {
            throw new IllegalArgumentException("recorded time range must be non-empty");
        }
    }
}

public record CaseStatusAssertion(
        AssertionId assertionId,
        CaseId caseId,
        CaseStatus status,
        ValidTimeRange validTime,
        RecordedTimeRange recordedTime,
        SourceEventId sourceEventId,
        Optional<AssertionId> supersedesAssertionId,
        Optional<CorrectionReason> correctionReason,
        OutputLineage lineage
) {}

public record CorrectionCommand(
        CaseId caseId,
        AssertionId assertionToCorrect,
        CorrectionReason reason,
        ValidTimeRange correctedValidTime,
        CaseStatus correctedStatus,
        Instant recordedAt,
        SourceEventId sourceEventId
) {}

1. find currently recorded assertion being corrected
2. verify correction is authorized and causally valid
3. close old assertion's recorded interval
4. insert corrected assertion with new recorded_from
5. update projection
6. emit correction lineage if needed

public void applyCorrection(CorrectionCommand command) {
    Instant recordedAt = command.recordedAt();

    CaseStatusAssertion old = repository.getOpenRecordedAssertion(
        command.assertionToCorrect()
    );

    CaseStatusAssertion closedOld = old.withRecordedTo(recordedAt);

    CaseStatusAssertion corrected = new CaseStatusAssertion(
        AssertionId.newDeterministic(command.sourceEventId()),
        command.caseId(),
        command.correctedStatus(),
        command.correctedValidTime(),
        new RecordedTimeRange(recordedAt, TimeConstants.INFINITY),
        command.sourceEventId(),
        Optional.of(old.assertionId()),
        Optional.of(command.reason()),
        currentLineage()
    );

    repository.transaction(() -> {
        repository.closeRecordedInterval(closedOld);
        repository.insert(corrected);
        projection.apply(corrected);
    });
}

For each case_id and recorded-time view, valid intervals for primary status must not overlap.

select a.case_id, a.assertion_id, b.assertion_id
from case_status_bitemporal a
join case_status_bitemporal b
  on a.case_id = b.case_id
 and a.assertion_id <> b.assertion_id
 and a.recorded_to = timestamp '9999-12-31 00:00:00'
 and b.recorded_to = timestamp '9999-12-31 00:00:00'
 and a.valid_from < b.valid_to
 and b.valid_from < a.valid_to;

Case reopened after closure

The previous assertion should no longer be considered valid truth.

{
  "eventType": "CaseStatusRetracted",
  "caseId": "C-100",
  "retractsAssertionId": "assertion-777",
  "reason": "SOURCE_ENTRY_ERROR",
  "recordedAt": "2026-04-10T09:00:00Z"
}

March 2026 enforcement SLA report is restated after correction batch.

{
  "restatementId": "rst-2026-04-sla-001",
  "supersedesReportRunId": "report-2026-04-05-001",
  "reason": "Late effective-date corrections received on 2026-04-10",
  "validPeriod": "2026-03",
  "recordedAsOf": "2026-04-12T00:00:00Z",
  "producedByRunId": "bf-2026-04-12-sla-restatement-001"
}

This newer output supersedes that older output.

parse -> classify -> derive assertion -> check duplicates -> resolve correction -> write ledger -> update projections -> validate

Effective date changes from April 1 to March 30.

public interface CorrectionImpactAnalyzer<E> {
    Set<OutputPartition> impactedPartitions(E correction);
}

public Set<OutputPartition> impactedPartitions(CaseStatusCorrection correction) {
    LocalDate oldDate = correction.oldValidFrom().atZone(zone).toLocalDate();
    LocalDate newDate = correction.newValidFrom().atZone(zone).toLocalDate();

    return DateRange.closed(min(oldDate, newDate), max(oldDate, newDate).plusDays(1))
            .stream()
            .flatMap(date -> Stream.of(
                    OutputPartition.daily(date),
                    OutputPartition.monthly(YearMonth.from(date))
            ))
            .collect(Collectors.toSet());
}

Case status joined to jurisdiction calendar.

case.valid_from >= calendar.valid_from
and case.valid_from < calendar.valid_to
and report.recorded_as_of >= calendar.recorded_from
and report.recorded_as_of < calendar.recorded_to

public enum TruthMode {
    CURRENT_ACCEPTED,
    AS_REPORTED,
    AS_KNOWN_AT,
    AS_EFFECTIVE_AT,
    REVISED_TRUTH,
    SOURCE_OBSERVED
}

create view current_case_status as
select *
from case_status_bitemporal s
where s.recorded_to = timestamp '9999-12-31 00:00:00'
  and s.valid_from <= current_timestamp
  and current_timestamp < s.valid_to;

where s.valid_from <= :valid_as_of
  and :valid_as_of < s.valid_to
  and s.recorded_from <= :recorded_as_of
  and :recorded_as_of < s.recorded_to

valid_from
valid_to
recorded_from
recorded_to
assertion_id
supersedes_assertion_id

What files/rows were in the table at snapshot N?

What business facts were valid at time Y and known at time X?

case-events-v1

case-events-v1
case-corrections-v1

case-status-assertions-v1

Partition by business key when ordering corrections relative to original assertions matters.

create table pending_correction (
    correction_event_id varchar primary key,
    target_assertion_id varchar not null,
    case_id varchar not null,
    payload jsonb not null,
    first_seen_at timestamp not null,
    retry_after timestamp not null,
    status varchar not null
);

C-100 effective date corrected
C-100 status reason corrected

March +1

April -1
March +1

create table aggregate_contribution_ledger (
    contribution_id varchar primary key,
    aggregate_name varchar not null,
    aggregate_key varchar not null,
    source_assertion_id varchar not null,
    contribution_value decimal not null,
    recorded_from timestamp not null,
    recorded_to timestamp not null,
    produced_by_run_id varchar not null
);

OPEN -> INVESTIGATING -> CLOSED

ledger of assertions -> sort by valid time -> replay domain state machine -> produce versioned projection

public final class CaseLifecycleRebuilder {
    public CaseLifecycleProjection rebuild(
            CaseId caseId,
            List<CaseLifecycleAssertion> assertions,
            TruthMode truthMode,
            Instant validAsOf,
            Instant recordedAsOf
    ) {
        List<CaseLifecycleAssertion> visible = assertions.stream()
                .filter(a -> visibleUnder(a, truthMode, validAsOf, recordedAsOf))
                .sorted(Comparator
                        .comparing((CaseLifecycleAssertion a) -> a.validTime().fromInclusive())
                        .thenComparing(a -> a.recordedTime().fromInclusive())
                        .thenComparing(a -> a.assertionId().value()))
                .toList();

        CaseLifecycleState state = CaseLifecycleState.initial(caseId);
        for (CaseLifecycleAssertion assertion : visible) {
            state = state.apply(assertion);
        }
        return state.toProjection();
    }
}

public final class BitemporalValidator {
    public List<Violation> validate(CaseStatusAssertion assertion) {
        List<Violation> violations = new ArrayList<>();

        if (!assertion.validTime().fromInclusive().isBefore(assertion.validTime().toExclusive())) {
            violations.add(new Violation("VALID_TIME_EMPTY"));
        }

        if (!assertion.recordedTime().fromInclusive().isBefore(assertion.recordedTime().toExclusive())) {
            violations.add(new Violation("RECORDED_TIME_EMPTY"));
        }

        if (assertion.sourceEventId() == null) {
            violations.add(new Violation("MISSING_SOURCE_EVENT_ID"));
        }

        return violations;
    }
}

{
  "runId": "bf-2026-04-sla-restatement-001",
  "truthMode": "REVISED_TRUTH",
  "validRange": {
    "from": "2026-03-01",
    "to": "2026-04-01"
  },
  "recordedAsOf": "2026-04-12T00:00:00Z"
}

create table regulatory_report_case_sla (
    report_run_id varchar not null,
    report_period varchar not null,
    jurisdiction varchar not null,
    breach_count bigint not null,
    open_case_count bigint not null,
    valid_from date not null,
    valid_to date not null,
    recorded_as_of timestamp not null,
    produced_by_run_id varchar not null,
    supersedes_report_run_id varchar null,
    primary key (report_run_id, jurisdiction)
);

Show me March report as filed on April 5.
Show me March report restated on April 12.
Show me why they differ.

{
  "oldReportRunId": "report-2026-04-05-001",
  "newReportRunId": "report-2026-04-12-001",
  "period": "2026-03",
  "differences": [
    {
      "metric": "sla_breach_count",
      "jurisdiction": "JKT",
      "oldValue": 182,
      "newValue": 189,
      "delta": 7
    }
  ],
  "causes": [
    {
      "correctionReason": "EFFECTIVE_DATE_CORRECTION",
      "count": 9
    },
    {
      "correctionReason": "LATE_CASE_CLOSURE",
      "count": 3
    }
  ]
}

@Test
void queryReturnsOldBeliefBeforeCorrectionRecorded() {
    var oldAssertion = assertion(
            validFrom("2026-03-31"),
            recordedFrom("2026-04-02"),
            recordedTo("2026-04-10")
    );

    var correctedAssertion = assertion(
            validFrom("2026-03-30"),
            recordedFrom("2026-04-10"),
            recordedTo(INFINITY)
    );

    var result = query.asKnownAt(
            caseId("C-100"),
            validAt("2026-03-31"),
            recordedAt("2026-04-05")
    );

    assertEquals(oldAssertion, result);
}

CaseOpened
CaseAssigned
CaseEscalated
SlaPaused
SlaResumed
CaseDecisionIssued
CaseClosed
CaseStatusCorrected
CaseEffectiveDateCorrected

Reports are not overwritten silently. They are superseded by restatements with explicit recorded_as_of and reason.

This is the value now.

This is what we believed then, about what was effective then, produced by this run, based on these source assertions, later superseded by this correction for this reason.