Zustand aus Events rekonstruieren

Was bedeutet den Zustand neu aufzubauen?

Eine einfache Ereignistabelle

CREATE TABLE account_events (
  event_id   SERIAL PRIMARY KEY,
  account_id INT NOT NULL,
  event_type VARCHAR(20) NOT NULL,  -- 'deposit' or 'withdrawal'
  amount     NUMERIC(12, 2) NOT NULL,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

INSERT INTO account_events (account_id, event_type, amount, created_at) VALUES
  (1, 'deposit',    1000.00, '2024-01-01 09:00:00+00'),
  (1, 'deposit',     500.00, '2024-01-03 14:00:00+00'),
  (1, 'withdrawal',  200.00, '2024-01-05 10:00:00+00'),
  (1, 'deposit',     300.00, '2024-01-07 11:00:00+00'),
  (1, 'withdrawal',  150.00, '2024-01-09 16:00:00+00');

Ereignisse zu einem Saldo zusammenführen

SELECT
  account_id,
  SUM(
    CASE event_type
      WHEN 'deposit'    THEN  amount
      WHEN 'withdrawal' THEN -amount
      ELSE 0
    END
  ) AS current_balance
FROM account_events
WHERE account_id = 1
GROUP BY account_id;

Zustand zu einem bestimmten Zeitpunkt

-- What was the balance at the end of January 5th?
SELECT
  account_id,
  SUM(
    CASE event_type
      WHEN 'deposit'    THEN  amount
      WHEN 'withdrawal' THEN -amount
      ELSE 0
    END
  ) AS balance_at_snapshot
FROM account_events
WHERE account_id = 1
  AND created_at <= '2024-01-05 23:59:59+00'
GROUP BY account_id;

Laufender Saldo mit Fensterfunktionen

SELECT
  event_id,
  created_at,
  event_type,
  amount,
  SUM(
    CASE event_type
      WHEN 'deposit'    THEN  amount
      WHEN 'withdrawal' THEN -amount
      ELSE 0
    END
  ) OVER (PARTITION BY account_id ORDER BY created_at, event_id)
    AS running_balance
FROM account_events
WHERE account_id = 1
ORDER BY created_at, event_id;

Zustand in einer Snapshot-Tabelle materialisieren

CREATE TABLE account_snapshots (
  account_id      INT PRIMARY KEY,
  current_balance NUMERIC(12, 2) NOT NULL,
  as_of_event_id  INT NOT NULL,
  updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Populate / refresh the snapshot from the event log
INSERT INTO account_snapshots (account_id, current_balance, as_of_event_id, updated_at)
SELECT
  account_id,
  SUM(CASE event_type WHEN 'deposit' THEN amount WHEN 'withdrawal' THEN -amount ELSE 0 END),
  MAX(event_id),
  NOW()
FROM account_events
GROUP BY account_id
ON CONFLICT (account_id) DO UPDATE
  SET current_balance = EXCLUDED.current_balance,
      as_of_event_id  = EXCLUDED.as_of_event_id,
      updated_at      = EXCLUDED.updated_at;

Inkrementelle Snapshot-Aktualisierungen

-- Apply only new events since the last snapshot
UPDATE account_snapshots AS snap
SET
  current_balance = snap.current_balance + delta.net,
  as_of_event_id  = delta.max_event_id,
  updated_at      = NOW()
FROM (
  SELECT
    ae.account_id,
    SUM(CASE ae.event_type WHEN 'deposit' THEN ae.amount WHEN 'withdrawal' THEN -ae.amount ELSE 0 END) AS net,
    MAX(ae.event_id) AS max_event_id
  FROM account_events ae
  JOIN account_snapshots s ON s.account_id = ae.account_id
  WHERE ae.event_id > s.as_of_event_id
  GROUP BY ae.account_id
) AS delta
WHERE snap.account_id = delta.account_id;

Temporale Tabellen und Systemversionierung

-- Emulating a temporal table in PostgreSQL
CREATE TABLE account_state_history (
  account_id      INT NOT NULL,
  current_balance NUMERIC(12, 2) NOT NULL,
  valid_from      TIMESTAMPTZ NOT NULL,
  valid_to        TIMESTAMPTZ NOT NULL DEFAULT 'infinity'
);

-- Insert initial state
INSERT INTO account_state_history (account_id, current_balance, valid_from)
VALUES (1, 1000.00, '2024-01-01 09:00:00+00');

-- On update: close old row, insert new row
UPDATE account_state_history
  SET valid_to = '2024-01-03 14:00:00+00'
WHERE account_id = 1 AND valid_to = 'infinity';

INSERT INTO account_state_history (account_id, current_balance, valid_from)
VALUES (1, 1500.00, '2024-01-03 14:00:00+00');

Temporale Historie abfragen

-- What was the account balance on January 4th?
SELECT
  account_id,
  current_balance,
  valid_from,
  valid_to
FROM account_state_history
WHERE account_id = 1
  AND valid_from <= '2024-01-04 00:00:00+00'
  AND valid_to   >  '2024-01-04 00:00:00+00';

Event Sourcing mit mehreren Entitäten

CREATE TABLE inventory_events (
  event_id    SERIAL PRIMARY KEY,
  product_id  INT NOT NULL,
  event_type  VARCHAR(20) NOT NULL,  -- 'received', 'shipped', 'adjusted'
  quantity    INT NOT NULL,
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

INSERT INTO inventory_events (product_id, event_type, quantity, created_at) VALUES
  (101, 'received',  200, '2024-03-01 08:00:00+00'),
  (101, 'shipped',    50, '2024-03-02 12:00:00+00'),
  (101, 'shipped',    30, '2024-03-04 15:00:00+00'),
  (102, 'received',  150, '2024-03-01 08:00:00+00'),
  (102, 'adjusted',  -10, '2024-03-03 09:00:00+00');

-- Rebuild current stock for all products
SELECT
  product_id,
  SUM(CASE event_type WHEN 'received' THEN quantity WHEN 'shipped' THEN -quantity ELSE quantity END) AS stock_on_hand
FROM inventory_events
GROUP BY product_id
ORDER BY product_id;

CTEs für mehr Übersichtlichkeit verwenden

CREATE TABLE accounts (
  account_id INT PRIMARY KEY,
  owner_name VARCHAR(100) NOT NULL
);

INSERT INTO accounts (account_id, owner_name) VALUES
  (1, 'Alice'),
  (2, 'Bob');

INSERT INTO account_events (account_id, event_type, amount, created_at) VALUES
  (2, 'deposit',   2000.00, '2024-01-02 10:00:00+00'),
  (2, 'withdrawal', 400.00, '2024-01-06 11:00:00+00');

WITH rebuilt_balances AS (
  SELECT
    account_id,
    SUM(CASE event_type WHEN 'deposit' THEN amount WHEN 'withdrawal' THEN -amount ELSE 0 END) AS balance
  FROM account_events
  GROUP BY account_id
)
SELECT
  a.account_id,
  a.owner_name,
  rb.balance
FROM accounts a
JOIN rebuilt_balances rb USING (account_id)
ORDER BY a.account_id;

Wissenscheck

Zusammenfassung der Lektion