Ticket Import Workflow

This example demonstrates a multi-step Causet saga for importing concert ticket data. The saga tracks progress through parsing, artist matching, metadata enrichment, and final storage — with explicit failure states at each step.

Domain

Users import ticket confirmation emails or PDF files. Causet coordinates the multi-step process: extract data, match to known artists and venues, enrich with metadata, store the ticket, and create a concert memory.

Entities and events

# states/ticket_import.state.causet
state:
  ticket_import:
    entity_key: import_id
    fields:
      - name: user_id
        type: string
        default: ""
      - name: raw_data
        type: string
        default: ""
      - name: artist_id
        type: string
        default: ""
      - name: venue_id
        type: string
        default: ""
      - name: show_date
        type: string
        default: ""
      - name: _tmp/saga_step
        type: int
        default: 0
      - name: _tmp/error_code
        type: string
        default: ""

# events/ticket_import.events.causet
events:
  TICKET_IMPORT_STARTED:
    state: ticket_import
    entity_expr: event.import_id
    payload:
      import_id: string
      user_id:   string
      raw_data:  string
 
  TICKET_PARSE_COMPLETED:
    state: ticket_import
    entity_expr: event.import_id
    payload:
      import_id:  string
      artist_name: string
      venue_name:  string
      show_date:   string
 
  TICKET_ARTIST_MATCHED:
    state: ticket_import
    entity_expr: event.import_id
    payload:
      import_id: string
      artist_id: string
 
  TICKET_IMPORT_COMPLETE:
    state: ticket_import
    entity_expr: event.import_id
    payload:
      import_id: string
      show_id:   string
 
  TICKET_IMPORT_FAILED:
    state: ticket_import
    entity_expr: event.import_id
    payload:
      import_id:  string
      error_code: string
      error_msg:  string

Saga definition

# sagas/ticket_import.sagas.causet
sagas:
  ticket_import_flow:
    state: ticket_import
    state_path: _tmp/saga_step
    steps:
      - name: idle
        set: { step: 0 }
      - name: started
        on: TICKET_IMPORT_STARTED
        set: { step: 1 }
      - name: parsed
        on: TICKET_PARSE_COMPLETED
        set: { step: 2 }
      - name: artist_matched
        on: TICKET_ARTIST_MATCHED
        set: { step: 3 }
      - name: complete
        on: TICKET_IMPORT_COMPLETE
        set: { step: 4 }
        end: true
      - name: failed
        on: TICKET_IMPORT_FAILED
        set: { step: -1 }
        end: true

Actions

# actions/ticket_import.actions.causet
actions:
  START_TICKET_IMPORT:
    state: ticket_import
    input:
      import_id: { type: string, required: true }
      user_id:   { type: string, required: true }
      raw_data:  { type: string, required: true }
    core:
      rules:
        - name: store_raw_data
          when: {}
          then:
            - op: set
              path: /user_id
              value: intent.user_id
            - op: set
              path: /raw_data
              value: intent.raw_data
    side_effects:
      rules:
        - name: emit_started
          then:
            - op: emit
              event_type: TICKET_IMPORT_STARTED
              payload:
                import_id: intent.import_id
                user_id:   intent.user_id
                raw_data:  intent.raw_data
 
  RECORD_TICKET_PARSE_RESULT:
    state: ticket_import
    input:
      import_id:   { type: string, required: true }
      artist_name: { type: string, required: true }
      venue_name:  { type: string, required: true }
      show_date:   { type: string, required: true }
    preflight:
      rules:
        - name: check_import_in_progress
          when: { expr: "entity._tmp.saga_step != 1" }
          then:
            - op: reject
              code: IMPORT_NOT_IN_STARTED_STATE
    side_effects:
      rules:
        - name: emit_parse_completed
          then:
            - op: emit
              event_type: TICKET_PARSE_COMPLETED
              payload:
                import_id:   intent.import_id
                artist_name: intent.artist_name
                venue_name:  intent.venue_name
                show_date:   intent.show_date
 
  MATCH_TICKET_ARTIST:
    state: ticket_import
    input:
      import_id: { type: string, required: true }
      artist_id: { type: string, required: true }
    preflight:
      rules:
        - name: check_import_parsed
          when: { expr: "entity._tmp.saga_step != 2" }
          then:
            - op: reject
              code: IMPORT_NOT_IN_PARSED_STATE
    core:
      rules:
        - name: store_artist_id
          when: {}
          then:
            - op: set
              path: /artist_id
              value: intent.artist_id
    side_effects:
      rules:
        - name: emit_artist_matched
          then:
            - op: emit
              event_type: TICKET_ARTIST_MATCHED
              payload:
                import_id: intent.import_id
                artist_id: intent.artist_id
 
  COMPLETE_TICKET_IMPORT:
    state: ticket_import
    input:
      import_id: { type: string, required: true }
      show_id:   { type: string, required: true }
    side_effects:
      rules:
        - name: emit_complete
          then:
            - op: emit
              event_type: TICKET_IMPORT_COMPLETE
              payload:
                import_id: intent.import_id
                show_id:   intent.show_id
 
  FAIL_TICKET_IMPORT:
    state: ticket_import
    input:
      import_id:  { type: string, required: true }
      error_code: { type: string, required: true }
      error_msg:  { type: string, required: true }
    core:
      rules:
        - name: store_error
          when: {}
          then:
            - op: set
              path: /_tmp/error_code
              value: intent.error_code
    side_effects:
      rules:
        - name: emit_failed
          then:
            - op: emit
              event_type: TICKET_IMPORT_FAILED
              payload:
                import_id:  intent.import_id
                error_code: intent.error_code
                error_msg:  intent.error_msg

Projection

# projections/ticket_import.projections.causet
projections:
  ticket_import_status:
    source_events: [
      TICKET_IMPORT_STARTED,
      TICKET_PARSE_COMPLETED,
      TICKET_ARTIST_MATCHED,
      TICKET_IMPORT_COMPLETE,
      TICKET_IMPORT_FAILED
    ]
    target:
      table: ticket_import_status
      primary_key: [import_id]
    fields:
      import_id:   TEXT
      user_id:     TEXT
      status:      TEXT
      artist_name: TEXT
      artist_id:   TEXT
      show_id:     TEXT
      error_code:  TEXT
      started_at:  BIGINT
      completed_at: BIGINT
      updated_at:  BIGINT
    derive:
      import_id:   event.import_id
      updated_at:  event.ts
    mutations:
      TICKET_IMPORT_STARTED:
        op: upsert
      TICKET_PARSE_COMPLETED:
        op: upsert
      TICKET_ARTIST_MATCHED:
        op: upsert
      TICKET_IMPORT_COMPLETE:
        op: upsert
      TICKET_IMPORT_FAILED:
        op: upsert
    indexes:
      - columns: [user_id]
      - columns: [status]

Query

# queries/ticket_import.queries.causet
queries:
  imports_for_user:
    from: ticket_import_status
    input:
      user_id: { type: string, required: true }
    where:
      user_id: { eq: input.user_id }
    order_by:
      started_at: desc
    limit: 50

Why sagas here

The ticket import flow spans multiple external service calls and can fail at any step. The saga provides:

Durability: saga state survives restarts (it’s in entity snapshot)
Observability: current step visible in projection
Explicit failure states: no silent half-completed imports
Replay safety: saga progression is driven by events, which are replayable

How external processing integrates

External processing services (parsers, AI models, metadata APIs) are NOT part of the Causet rules. They integrate by submitting intents through an SDK:

External parser service:
1. Picks up TICKET_IMPORT_STARTED event from causet.ledger-events.v1 Kafka topic
2. Parses the raw_data
3. Calls RECORD_TICKET_PARSE_RESULT via causet-sdk (Python example below)

This keeps rules deterministic (no external I/O) while integrating with real-world services.

Testing the flow

Use the Python SDK to drive the saga end to end:

import asyncio
from causet_sdk import CausetClient
 
async def main():
    client = CausetClient(
        api_url="http://localhost:8085",
        platform_slug="jamlet",
        app_slug="concert-app",
        fork_id="main",
        api_key="ck_live_xxx.secret",
    )
    await client.init()
 
    # Step 1: Start import
    await client.emit(
        "ticket_import_stream", "import-1", "START_TICKET_IMPORT",
        {
            "import_id": "import-1",
            "user_id": "user-1",
            "raw_data": "Pearl Jam - Sep 15 - Barclays Center",
        },
    )
 
    # Step 2: External parser calls back (simulate parser service)
    await client.emit(
        "ticket_import_stream", "import-1", "RECORD_TICKET_PARSE_RESULT",
        {
            "import_id": "import-1",
            "artist_name": "Pearl Jam",
            "venue_name": "Barclays Center",
            "show_date": "2026-09-15",
        },
    )
 
    # Step 3: Artist matching
    await client.emit(
        "ticket_import_stream", "import-1", "MATCH_TICKET_ARTIST",
        {"import_id": "import-1", "artist_id": "artist-pearl-jam"},
    )
 
    # Query status
    rows = await client.run_query("imports_for_user", {"user_id": "user-1"})
    print(rows["items"])
 
    client.destroy()
 
asyncio.run(main())

Expected query response:

{
  "items": [
    {
      "import_id": "import-1",
      "user_id": "user-1",
      "status": "artist_matched",
      "artist_id": "artist-pearl-jam"
    }
  ]
}

Support Copilot (AI)User Notifications