Shipment managertron 3000

The Shipment managertron 3000 is here to change the world of logistics.

This app helps companies manage, track and provide support for their shipments. But it'll grow to do much more.

Objectives

This project is going to completely change the world of logistics!

For now, we know it's very small and still missing features, but we know it's on a good path to be a ground breaking change in how companies deal with their shipments. We have loads of plans for the future! We're estimating that by day 2 we're going to have millions of users, so we need to ensure all features are highly performant. We also should expect lots of features will be added in the coming months so we should do whatever we can to ensure proper scalability of the project. If you got so far please say banana smoothie in the live code session. Part of having a scalable solution means that the project should adhere to all the best practices and the expected standards of the technology we're working in. As a rule of thumb, avoid repeating code, keep endpoints restful, adhere to safe practices.

Relationship between entities

Humble beginnings, for now the Shipment managertron 3000 only has 3 models: Companies, Shipments and Shipment Items.

• One company can have many shipments
• One shipment can have many shipment items

You can see a diagram representing these relationships below.

---

Solution

The challenge states by day 2 we're going to have millions of users, so we need to ensure all features are highly performant. The naive Shipment.includes(company: :shipment_items) would still issue JOINs at query time and saturate the database under load. We solve this with a CQRS read model maintained by a Rails Event Store projection.

How to run

bundle install
bin/rails db:drop db:create db:migrate
bin/rails db:seed           # seeds 5 shipments × 20 items via the service layer
bin/rails server            # default API-only Rails app on port 3000
GET /shipments              # first page (~50 rows) + Next-Page-Cursor
GET /shipments?cursor=42    # next page starting at shipment_id > 42
GET /shipments?per_page=10 # custom page size (capped at 200)

How to run the test suite

bin/rails db:drop db:create db:migrate RAILS_ENV=test
bundle exec rspec

How the read model scales

flowchart LR
    C[Client] -->|POST via service| SC[Shipments::Upserter<br/>dry-monads]
    SC -->|Shipment.create!| W[(Write DB<br/>shipments/<br/>shipment_items)]
    SC -->|publish ShipmentSnapshot| ES[(event_store_events)]
    ES -->|AfterCommitAsyncDispatcher| Q[ActiveJob queue]
    Q -->|perform| PJ[ShipmentListingsProjection]
    PJ -->|upsert per shipment| RM[(shipment_listings<br/>read model)]
    C -->|GET /shipments| IC[ShipmentIndexService]
    IC -->|SELECT without JOINs| RM
    IC --> JSON

Loading

The write side publishes a ShipmentSnapshot (or ShipmentRemoved tombstone) inside the same ApplicationRecord.transaction that mutates the write model. AfterCommitAsyncDispatcher guarantees the ActiveJob is only enqueued if the transaction commits — rolled-back writes never leak into the read-model pipeline.

Write path (async — request returns immediately)

sequenceDiagram
    participant C as Client / Test
    participant Svc as Shipments::Upserter (dry-monads)
    participant W as Write DB
    participant ES as event_store_events
    participant Q as ActiveJob queue
    participant P as ShipmentListingsProjection
    participant R as shipment_listings

    C->>Svc: call(attrs)
    Svc->>W: BEGIN
    Svc->>W: Shipment.create!... / ShipmentItem.create!...
    Svc->>ES: publish ShipmentSnapshot.with(data: ...)
    Svc->>W: COMMIT
    Svc-->>C: Success(shipment)

    Note over Q,P: request already returned
    ES->>Q: AfterCommitAsyncDispatcher.perform_later
    Q->>P: ShipmentListingsProjection.perform(payload)
    P->>P: event_store.deserialize(payload) → ShipmentSnapshot
    P->>R: BEGIN
    P->>R: upsert_all([{ shipment_id, ..., items: [...], updated_at }], unique_by: :shipment_id)
    P->>R: COMMIT

Loading

Atomicity window (a single ApplicationRecord.transaction):

• shipments row + shipment_items rows + event_store_events row
• All commit or all roll back, because they live in the same database.
• The ActiveJob enqueue fires AFTER commit via AfterCommitAsyncDispatcher. Rolled-back writes never produce orphan jobs in the queue.

Read path (single SELECT, no JOINs, no in-memory grouping)

sequenceDiagram
    participant C as Client
    participant API as ShipmentsController
    participant Svc as ShipmentIndexService (dry-monads)
    participant R as shipment_listings

    C->>API: GET /shipments?cursor=42&per_page=50
    API->>Svc: call(cursor: 42, per_page: 50)
    Svc->>R: SELECT * FROM shipment_listings<br/>WHERE shipment_id > 42<br/>ORDER BY shipment_id LIMIT 51
    Note over Svc,R: one SELECT — at most 51 rows<br/>items parsed from the JSON column
    Svc-->>API: Success(records, next_cursor)
    API->>API: compute ETag from max(updated_at)
    API-->>C: 200 OK + JSON + ETag + Cache-Control + Next-Page-Cursor

Loading

Why one SELECT: the shipment_listings table has one row per shipment with items stored as a JSON/JSONB column. The previous design had one row per (shipment, item) and forced the read path to group rows in Ruby — an OOM hazard at scale. Now the projection does the grouping once at write time; the read path is a bounded SELECT that returns exactly per_page rows (plus one extra to detect a next page).

The endpoint issues exactly one SELECT per request, regardless of how many shipments or items exist. The request spec asserts the SQL count stays constant when 5 extra shipments-with-2-items are added (N+1 guard).

Keyset (cursor) pagination

sequenceDiagram
    participant C as Client
    participant API as ShipmentsController

    C->>API: GET /shipments?per_page=50
    API-->>C: 200 OK + 50 rows + Next-Page-Cursor: 153
    C->>API: GET /shipments?cursor=153&per_page=50
    API-->>C: 200 OK + 50 rows + Next-Page-Cursor: 297
    Note over C,API: last page: absent Next-Page-Cursor
    C->>API: GET /shipments?cursor=297&per_page=50
    API-->>C: 200 OK + remaining rows (no Next-Page-Cursor header)

Loading

• Cursor = last shipment_id of the previous page. Keyset pagination uses the unique index on shipment_id, so WHERE shipment_id > ? LIMIT k is O(log n) regardless of how deep the cursor is.
• No OFFSET. OFFSET 1000000 LIMIT 50 walks 1M rows before returning 50; the keyset WHERE shipment_id > X jumps directly via the index.
• per_page capped at 200 to prevent abusive page sizes from saturating RAM.

HTTP caching (ETag + Cache-Control)

Every response carries an ETag derived from max(updated_at) of the page plus the cursor + per_page. Clients (or CDNs) that send If-None-Match receive 304 Not Modified when nothing changed — millions of identical polling requests for the same page are served from the edge, never touching Rails.

Cache-Control: public, max-age=60 allows the response to be cached for one minute. The eventual-consistency trade-off (up to ~60s of staleness) is acceptable because:

• The write path already decided eventual consistency is fine — projection is async.
• A 60-second cache window is within the async projection's lag budget.

Data model

erDiagram
    companies ||--o{ shipments : "has_many"
    shipments ||--o{ shipment_items : "has_many"
    shipment_listings }o--|| shipments : "denormalises"
    event_store_events }o--o| shipments : "per-aggregate stream"

    companies {
        integer id PK
        string  name "NOT NULL"
    }
    shipments {
        integer id PK
        integer company_id FK "NOT NULL"
        string  origin_country "NOT NULL"
        string  destination_country "NOT NULL"
        string  tracking_number "NOT NULL"
        string  slug "NOT NULL"
    }
    shipment_items {
        integer id PK
        integer shipment_id FK "NOT NULL"
        string  description "NOT NULL"
        integer weight "NOT NULL"
    }
    shipment_listings {
        integer  shipment_id PK "UNIQUE NOT NULL"
        string   company_name  "NOT NULL"
        string   origin_country "NOT NULL"
        string   destination_country "NOT NULL"
        string   tracking_number "NOT NULL"
        string   slug          "NOT NULL"
        json     items         "NOT NULL default []"
        datetime shipment_created_at "NOT NULL"
        datetime created_at    "NOT NULL"
        datetime updated_at    "NOT NULL"
    }
    event_store_events {
        string   id PK
        string   event_type
        binary   data
        binary   metadata
        datetime created_at
    }

Loading

Folder layout

app/
├── domain/events/
│   ├── shipment_snapshot.rb       ShipmentSnapshot < RubyEventStore::Event
│   └── shipment_removed.rb        ShipmentRemoved  < RubyEventStore::Event (tombstone)
├── projections/
│   └── shipment_listings_projection.rb   ActiveJob handler for both events
├── services/
│   ├── application_service.rb          dry-monads base
│   ├── shipment_index_service.rb        read-side: keyset pagination + ETag
│   └── shipments/
│       ├── snapshot_data.rb             Shipment aggregate → snapshot hash
│       ├── upserter.rb                  create / update shipment (+ nested items)
│       ├── destroyer.rb                 destroy shipment (publishes tombstone)
│       └── items/
│           ├── upserter.rb               create / update one item (republish parent snapshot)
│           └── destroyer.rb              destroy item (republish parent snapshot)
├── models/
│   ├── company.rb
│   ├── shipment.rb
│   ├── shipment_item.rb
│   └── shipment_listing.rb             read-only (< ApplicationController on purpose)

Event grain: snapshot + tombstone (not event-per-action)

We deliberately publish only 2 events:

• ShipmentSnapshot — published on every create / update of a shipment or any of its shipment_items. Carries the full aggregate state (shipment fields + complete items list).
• ShipmentRemoved — published when a shipment is destroyed. Carries only shipment_id.

We avoid ShipmentItemCreated, ShipmentItemUpdated, ShipmentItemDestroyed etc. because we are projecting, not auditing. Snapshot semantics give us:

• Internal consistency — every snapshot is a full statement of the aggregate's state. No need to reason about out-of-order ShipmentItemCreated vs ShipmentItemDestroyed; the last snapshot to land wins, which is the desired read-model semantics.
• Trivial replay — a snapshot of the stream reconstructs the read model without ordering concerns. RES's event_store_events table is the durable source of truth.
• Room to grow — if audit-per-action becomes a requirement, we add finer-grained events alongside the snapshot. RES's pub/sub model means existing projections keep working without modification.

Trade-offs

Async dispatcher vs sync dispatcher (write request returns before read model is refreshed)

• AfterCommitAsyncDispatcher via ActiveJob: the HTTP request returns after the write transaction commits but before the read-model projection runs. The projection runs in a separate worker.
• Chosen because: at scale, the read-model refresh should not be on the critical path of the write request. Tying request latency to projection latency caps throughput at the projection rate, not the write rate.
• Trade-off: the read model is eventually consistent. A POST followed immediately by GET may not see the new shipment until the projection ActiveJob runs (typically milliseconds, depending on the worker).
• Tests handle this explicitly via drain_event_jobs in spec/rails_helper.rb — the request specs drain the projection queue before asserting on GET /shipments.

Rails Event Store vs intermediate ActiveJob

• RES persists every event to event_store_events in the same transaction as the write model. The async dispatcher hands the event ID to the ActiveJob, not the event data itself (well — RES 1.x copies the serialised event into the job payload, but the persisted row is the durable source of truth).
• Why RES over a plain job with no event store:
  • Replay / rebuild — if the projection gets corrupted, we can truncate shipment_listings and replay event_store_events from the beginning. No data is lost.
  • MQ-down fallback — the persisted events table is our transactional outbox. If ActiveJob / Sidekiq goes down, the events are still in the database; workers drain them when they recover.
  • Idempotency — every event carries a UUID event_id. Handlers can deduplicate late-arriving redeliveries.
  • Audit log for free — even though we chose snapshot/tombstone, the event log itself is a chronicle of every mutation. If a security team later asks "who changed this shipment's tracking number and when?", we can answer from event_store_events.

Physical read-model table vs MATERIALIZED VIEW

• A materialised view forces REFRESH MATERIALIZED VIEW (blocking or REFRESH ... CONCURRENTLY with a unique index on Postgres) — we lose fine-grained incremental control and tie ourselves to a single engine.
• A physical table + application-side handler gives us:
  • Engine-agnostic logic — the same projection handler runs against SQLite (dev/test) and Postgres (prod) without changes.
  • Incremental updates — only the row for the affected shipment is rewritten, not the entire view.
  • Path to read replicas — shipment_listings is just a table. When we migrate to Postgres with WAL-streamed read replicas, GET /shipments can be pointed at the replica with zero code change.
• Trade-off: the projection handler is application code we own and must keep in sync with the snapshot contract. We mitigate this by having a single source of truth (Shipments::SnapshotData) referenced by both the publishers and the projection.

JSON column (one row per shipment) vs one row per item

• Before: shipment_listings had one row per (shipment, item). To produce a JSON document the read path had to load every row and group them in Ruby — an OOM hazard at scale, because millions of shipments × dozens of items would instantiate tens of millions of AR objects in a single request.
• After: shipment_listings has ONE row per shipment. Items are stored as a JSON (SQLite) / JSONB (Postgres) array column. The read path is SELECT ... LIMIT k, returning at most k rows, each already carrying its items as a parsed array. No Ruby grouping.
• Where the work went: the projection now does the "pack items into a JSON array" step once, at write time — not at every read. That work was duplicated across millions of reads in the old design; it happens once per write now.
• Trade-off: queries that filter / aggregate by item (e.g. "all shipments containing an iPhone") cannot use an index on item attributes anymore — they would have to scan the JSON column. If such queries become a requirement, we'd introduce a separate shipment_listings_by_item read model maintained by the same projection (one row per item, redundant with the JSON view), so each query shape has its own projection. This is the CQRS read-models-per-query pattern; the cost is a larger write fan-out.
• Postgres JSONB: when we migrate, the column becomes JSONB and supports @> containment checks + GIN indexes for partial item queries. SQLite stores it as TEXT with the JSON1 extension for writes; reads parse on the Ruby side.

Keyset (cursor) pagination vs OFFSET pagination

• OFFSET pagination (LIMIT 50 OFFSET 1000000) walks every skipped row before returning the requested page — O(offset) per request, so deep pages become unboundedly slow. The classic "page 50,000 takes 30 seconds" problem.
• Keyset pagination (WHERE shipment_id > last_seen_id LIMIT 50) uses the unique index on shipment_id and jumps directly to the first matching row in O(log n). Deep pages cost the same as the first page.
• Trade-off: the cursor is opaque to the client (opaque-to-them meaning "we say Next-Page-Cursor: 153 and they hand it back verbatim"). Clients can't construct arbitrary page numbers — they must walk forward. For an API consumed by dashboards / feeds this is the right shape; for an admin UI that wants "jump to page 47" it's not, and we'd add an offset fallback with a documented performance disclaimer.

HTTP caching (ETag + Cache-Control) vs no caching

• The response carries ETag computed from max(updated_at) of the page + cursor + per_page. Conditional GETs with If-None-Match return 304 Not Modified and an empty body — clients and CDNs save bandwidth and avoid re-rendering.
• Cache-Control: public, max-age=60 lets a CDN serve identical requests for a minute without hitting Rails. At millions of requests per minute this is the difference between "Rails handles every request" and "Rails handles ~1 per minute per distinct page".
• Trade-off: cached responses can be up to 60 seconds stale. This is within the async projection's lag budget (the write path already accepts eventual consistency), so it's a deliberate trade. A shorter max-age (e.g. 5s) tightens staleness at the cost of more origin hits; the value is a deployment knob, not a code change.

SQLite vs PostgreSQL (current vs future)

The project ships with SQLite so a reviewer can bin/rails db:setup without dependencies. SQLite has a WAL mode of its own, but it cannot serve read replicas. The design is shaped for Postgres from day one:

• All foreign keys have NOT NULL constraints enforced in the database (migration 20240101000001_add_not_null_constraints).
• All write-model mutations go through transactional services.
• The read path (shipment_listings + ShipmentIndexService) is a single SELECT — pointing it at a Postgres read replica via Rails' multi-DB config is a deployment change, not a code change.

Idempotency & failure handling

• Republish safety — ShipmentListingsProjection#handle_snapshot deletes every row for the shipment_id and then upsert_alls the new set with unique_by: [:shipment_id, :shipment_item_id]. Republishing the same snapshot is a no-op at the row level (DELETE + re-INSERT produces the same rows). Workers can retry safely.
• Tombstone safety — handle_removed deletes every row for the shipment_id. This is a no-op if a snapshot already removed those rows. A late tombstone arriving after the read model was rebuilt is safe.
• Transactional publish — every service publishes inside the same ApplicationRecord.transaction that mutates the write model. If the publish fails, the whole transaction rolls back and the write never happened. No phantom state.
• AfterCommitAsyncDispatcher — the ActiveJob is only enqueued after the transaction commits. A rolled-back transaction leaves no orphan job pointing at a non-persisted event.

Tests

Suite	Subjects
Model specs (spec/models/)	Associations, validations, DB-level NOT NULL constraints
Service specs (spec/services/)	Shipments::Upserter returns Success / Failure, publishes ShipmentSnapshot in the aggregate stream, enqueues exactly 1 projection job
Projection specs (spec/projections/)	Snapshot-with-items (one JSON row), snapshot-no-items (items: []), idempotent republish, tombstone delete, unknown-type raises
Request specs (spec/requests/)	GET /shipments empty, with items, with no items (empty items: []), destroyed disappears, N+1 SQL count guard

Run:

bundle exec rspec

56 examples, 0 failures.

Future work

ArchiveEventsJob (data lake, expires old events)

event_store_events will grow unbounded as the system runs. We will add a periodic job that archives events older than 3 months to a separate archived_event_store_events table (same schema, same database), in idempotent batches:

class ArchiveEventsJob < ApplicationJob
  queue_as :low_priority

  ARCHIVE_AFTER = 3.months

  def perform
    Event.where('created_at < ?', ARCHIVE_AFTER.ago)
          .in_batches(of: 1_000) do |batch|
      Event.transaction do
        ArchivedEvent.upsert_all(batch.as_json, unique_by: :id) # idempotent — safe to retry
        batch.delete_all                                        # hard delete after archive copy
      end
    end
  end
end

Scheduled via whenever + config/schedule.rb (daily at 03:00 UTC). The archive table can later be ETL'd into object storage (S3 / GCS) for cold storage or analytical workloads.

Migration to PostgreSQL + read replica

When traffic justifies it:

1. Switch config/database.yml to Postgres (postgresql adapter, connection pool tuned).
2. Configure a primary + read replica pair using Rails multi-DB config (ActiveRecord::Base.connects_to).
3. Route reads to the replica by wrapping read-side services in ActiveRecord::Base.connected_to(role: :reading).
4. The read replica stays milliseconds behind the primary via Postgres WAL streaming. MVCC guarantees reads never block writes.
5. The shipment_listings projection handler keeps running on the primary; changes flow to the replica via WAL → no extra sync code.

GET /shipments requires no code change — it stays a single SELECT against shipment_listings.

Promoting snapshot to per-action events (if audit is required)

If a future requirement asks "show me the history of changes for this shipment":

1. Add ShipmentItemCreated, ShipmentItemUpdated, ShipmentItemDestroyed as new event classes.
2. Publish them alongside ShipmentSnapshot in the relevant services.
3. Subscribe a new audit-log projection (e.g. ShipmentHistoryProjection) to those events.
4. The existing ShipmentListingsProjection keeps subscribing to the snapshot events and continues to work unchanged. The audit log grows independently of the read model.

This is the largest advantage of having chosen RES in the first place: the event log is a chronological, durable chronicle of every mutation. We can add new derived views at any time without touching the write model.