Approved, then “pending” again

/** * Build v1: check the signature first, store each message once, and apply messages in the provider's * version order. A message that arrives before an earlier version is parked, and applied as soon as * the gap closes. One applicant's messages never interleave (a row lock in Postgres). A refused * message is logged as evidence, apart from the stored ones (so it can never use up a real message's ID) * and outside the applicant's lock. */export const handleEvent: HandleEvent = async ({ store, signatures, refusals }, event) => {  const signatureValid = await signatures.verify(event)  if (!signatureValid) {    await refusals.record(event, EVENT_RESULT.BAD_SIGNATURE)    return EVENT_RESULT.BAD_SIGNATURE  }  const result = await store.forApplicant(event.applicantId, (tx) => applyInOrder(tx, event))  if (result === EVENT_RESULT.VERSION_CONFLICT) await refusals.record(event, result)  return result} /** Inside the applicant's lock: keep the message once, then apply it and every parked one that follows on. */async function applyInOrder(tx: ApplicantTx, event: ProviderEvent): Promise<EventResult> {  const recorded = await tx.recordEvent(event)  if (recorded === RECORD_RESULT.SAME_ID) return EVENT_RESULT.DUPLICATE  if (recorded === RECORD_RESULT.SAME_VERSION) {    // The same version under a new ID: a copy if it says the same, a conflict if it says something else.    const kept = await tx.storedEvent(event.version)    const sameContent = kept?.status === event.status    if (!sameContent) return EVENT_RESULT.VERSION_CONFLICT  }  const applicant = await tx.loadApplicant()  if (event.version <= applicant.version) return EVENT_RESULT.STALE  if (event.version > applicant.version + 1) return EVENT_RESULT.PARKED   // Apply this message, then every parked one that now follows on.  const applied = { ...applicant, status: event.status, version: event.version, lastSentAt: event.sentAt }  const caughtUp = await applyFollowing(tx, applied)  await tx.saveApplicant(caughtUp)  return EVENT_RESULT.APPLIED} /** From this state on, applies every kept message that follows on, one version after another (not saved yet). */export async function applyFollowing(tx: ApplicantTx, from: Applicant): Promise<Applicant> {  let applicant = from  let next = await tx.storedEvent(applicant.version + 1)  while (next) {    applicant = { ...applicant, status: next.status, version: next.version, lastSentAt: next.sentAt }    next = await tx.storedEvent(applicant.version + 1)  }  return applicant}

// HandleEvent is build v1 in Go: check the signature first, keep each message once, and apply messages in the// provider's version order. A message that arrives before an earlier version is parked, and applied as soon as the// gap closes. One applicant's messages never interleave (a row lock in Postgres). A refused message is logged as// evidence, apart from the kept ones (so it can never use up a real message's ID) and outside the applicant's lock.func HandleEvent(ctx context.Context, store Store, signatures SignatureCheck, refusals RefusalLog, event ProviderEvent) (EventResult, error) {	signatureValid := signatures.Verify(event)	if !signatureValid {		err := refusals.Record(ctx, event, BadSignature)		return BadSignature, err	}	result, err := store.ForApplicant(ctx, event.ApplicantID, func(tx Tx) (EventResult, error) {		return applyInOrder(ctx, tx, event)	})	if err != nil {		return "", err	}	if result == VersionConflict {		err = refusals.Record(ctx, event, result)	}	return result, err} // applyInOrder runs inside the applicant's lock: keep the message once, then apply it and every parked one that// follows on.func applyInOrder(ctx context.Context, tx Tx, event ProviderEvent) (EventResult, error) {	recorded, err := tx.RecordEvent(ctx, event)	if err != nil {		return "", err	}	if recorded == RecordSameID {		return Duplicate, nil	}	if recorded == RecordSameVersion {		// The same version under a new ID: a copy if it says the same, a conflict if it says something else.		kept, err := tx.StoredEvent(ctx, event.Version)		if err != nil {			return "", err		}		sameContent := kept != nil && kept.Status == event.Status		if !sameContent {			return VersionConflict, nil		}	}	applicant, err := tx.LoadApplicant(ctx)	if err != nil {		return "", err	}	if event.Version <= applicant.Version {		return Stale, nil	}	if event.Version > applicant.Version+1 {		return Parked, nil	} 	// Apply this message, then every parked one that now follows on.	next := &event	for next != nil {		applicant.Status, applicant.Version, applicant.LastSentAt = next.Status, next.Version, next.SentAt		next, err = tx.StoredEvent(ctx, applicant.Version+1)		if err != nil {			return "", err		}	}	err = tx.SaveApplicant(ctx, applicant)	return Applied, err}

      // The applicant's row exists from the start of the check; a first message creates it if not. Then the row      // is locked, before the message is kept: every message for this applicant waits here until the one before      // it has committed, so two never interleave (and a copy waits on the lock, not on a half-written row).      await tx.query(        `INSERT INTO kyc_applicants (applicant_id, status) VALUES ($1, 'NOT_STARTED')         ON CONFLICT (applicant_id) DO NOTHING`,        [applicantId],      )      if (this.#lockRow) {        await tx.query('SELECT 1 FROM kyc_applicants WHERE applicant_id = $1 FOR UPDATE', [applicantId])      }

	// The applicant's row exists from the start of the check; a first message creates it if not. Then the row is	// locked, before the message is kept: every message for this applicant waits here until the one before it has	// committed (and a copy waits on the lock, not on a half-written row).	_, err = tx.Exec(ctx, `INSERT INTO kyc_applicants (applicant_id, status) VALUES ($1, 'NOT_STARTED')		ON CONFLICT (applicant_id) DO NOTHING`, applicantID)	if err != nil {		return "", err	}	if s.lockRow {		if _, err := tx.Exec(ctx, `SELECT 1 FROM kyc_applicants WHERE applicant_id = $1 FOR UPDATE`, applicantID); err != nil {			return "", err		}	}