Skip to main content

Job seeker engagement

Keep an employer and a job seeker aligned while the job seeker decides whether to accept an opportunity.

Product requirements​

The process accepts an employer ID, job seeker ID, and initial notes. It must:

  1. Store both participants, set the status to Initiated, record the update time, and preserve the notes.
  2. Start three branches: watch the overall deadline, manage reminders, and notify the external recruiting system that the process started.
  3. Send a reminder every five seconds in the example while the status remains Initiated. A production deployment would use a longer interval.
  4. Let the job seeker opt out of reminders without ending the decision process.
  5. Let the job seeker decline while the process is Initiated. A declined process remains open and can later be accepted.
  6. Let the job seeker accept while the process is Initiated or Declined. Acceptance stops the overall wait, notifies the external system, and completes the Flow with done.
  7. Complete with timeout if no acceptance arrives within 60 days.
  8. Expose the current status and notes through an RPC, and make status searchable through an indexed Attribute.

The overall branch waits for either the 60-day Timer or a message on CompleteProcess. The reminder branch independently waits for either a five-second Timer or a message on OptOutReminder. Workers can stop during either wait without losing the deadline, reminder schedule, or external events.

Potential failures and exceptional outcomes include:

  • No acceptance arrives before the deadline. This is an expected timeout result, not an infrastructure failure.
  • A decline is requested after the status has changed, or an acceptance is requested after completion. The RPC rejects the invalid transition without scheduling a notification Step.
  • Email or recruiting-system calls can fail transiently. Their adapters should surface those errors so Dex can retry the owning Step.
  • A dependency can commit a side effect before the Worker records success. Email and notification calls therefore need idempotency keys or equivalent deduplication in production.
  • Reminder opt-out stops only future reminders. The status and overall acceptance deadline remain active.

The Go, Java, Python, and TypeScript examples implement the full 60-day process. The Rust released-SDK example keeps the same end-to-end start, describe, accept, decline, opt-out, notification, and search endpoints, but uses one decision Channel with a 24-hour Timer so its integration path stays compact.

Flow design​

This interactive definition graph is generated from the runnable Python Flow.

Definition graph

EngagementFlow

Valid

python · examples/python/dex_examples/products/engagement/engagement_flow.py

The main topology fans out after initialization:

Initialize → ProcessTimeout ∥ Reminder ∥ NotifyExternalSystem

It combines these design patterns:

  • Static parallel Steps separate the overall deadline, reminder loop, and external notification. A slow branch does not block the others.
  • Durable reminder moves Reminder back to itself after each Timer fires.
  • Interruptible wait uses AnyOf so a Channel message can wake a Step before its Timer.
  • Responsive update lets Accept and Decline RPCs validate and change persisted state while the Flow is waiting.
  • Attribute match lets callers wait until durable state is visible.
  • Graceful completion lets the timeout branch produce either done or timeout while the other branches finish or become dead ends.

The reminder and overall timeout use separate Channels because opting out of email must not accept, decline, or finish the opportunity. Status, notes, and timestamps live in Attributes so every branch and RPC observes the same durable state.

Core implementation​

Each tab starts with the Flow signature and topology, then shows the durable waits that control completion and reminders. The linked runnable example contains the complete RPC handlers, request models, controller, and integration test.

class EngagementFlow(Flow[EngagementInput]):
    employer_id = Attribute("EmployerId", str)
    job_seeker_id = Attribute("JobSeekerId", str)
    engagement_status = Attribute(
        "EngagementStatus",
        Status,
        AttributeIndex(IndexType.KEYWORD, STATUS_SEARCH_KEY),
    )
    last_update_timestamp = Attribute("LastUpdateTimeMillis", int)
    notes = Attribute("notes", str)
    opt_out_reminder = Channel[None]("OptOutReminder", type(None))
    complete_process = Channel[None]("CompleteProcess", type(None))

    def __init__(self, service: MyDependencyService) -> None:
        self.service = service
        self.initialize = Initialize(self)
        self.process_timeout = ProcessTimeout(self)
        self.reminder = Reminder(self)
        self.notify_external_system = NotifyExternalSystem(self)

    def get_steps(self) -> StepList[EngagementInput]:
        return StepList.start_step(self.initialize).other_steps(
            self.process_timeout,
            self.reminder,
            self.notify_external_system,
        )

    def get_persistence_schema(self) -> PersistenceSchema:
        return PersistenceSchema.of(
            self.employer_id,
            self.job_seeker_id,
            self.engagement_status,
            self.last_update_timestamp,
            self.notes,
            self.opt_out_reminder,
            self.complete_process,
        )
class ProcessTimeout(Step[None]):
    def __init__(self, flow: "EngagementFlow") -> None:
        self.flow = flow

    def wait_for(self, context: Context, input: None) -> Wait:
        return Wait.any_of(
            Timer.by_duration(timedelta(days=60)),
            self.flow.complete_process.for_one(),
        )

    def execute(self, context: Context, input: None) -> StepDecision:
        description = self.flow.describe_engagement(context)
        result = "timeout"
        if description.current_status == Status.ACCEPTED:
            result = "done"
        self.flow.service.update_external_system(
            f"engagement from employer {description.employer_id} "
            f"to job seeker {description.job_seeker_id} "
            f"finished with status {description.current_status}"
        )
        return graceful_complete(result)


class Reminder(Step[None]):
    def __init__(self, flow: "EngagementFlow") -> None:
        self.flow = flow

    def wait_for(self, context: Context, input: None) -> Wait:
        return Wait.any_of(
            Timer.by_duration(timedelta(seconds=5)),
            self.flow.opt_out_reminder.for_one(),
        )

    def execute(self, context: Context, input: None) -> StepDecision:
        status = self.flow.engagement_status.get(context)
        if status is not Status.INITIATED:
            return dead_end()
        if self.flow.opt_out_reminder.results(context):
            self.flow.update_status(context, status, "user opted out of reminders")
            return dead_end()
        self.flow.service.send_email(
            self.flow.job_seeker_id.get(context),
            "Reminder: please respond",
            "Please respond to the engagement.",
        )
        return go_to(Reminder, None)

Example: examples/python/dex_examples/products/engagement/engagement_flow.py

Demo​

Run Dex and any language example server, then start Job seeker engagement from the examples playground. The demo below uses the Go example. It initializes an employer and job seeker, opts out of reminders, records a decline, then accepts the opportunity. The Flow finishes with done after the acceptance message wakes ProcessTimeout.

Select the image to inspect the original 2370 × 1050 capture.

Completed Job seeker engagement Step execution graph