Build reliable
software effortlessly

# Define a durable checkout workflow
@DBOS.workflow()
def checkout_workflow(items):
    # Step 1: Create the order
    order = create_order()

    # Step 2: Reserve inventory for the items
    reserve_inventory(order, items)

    # Step 3: Process the payment
    payment_status = process_payment(order, items)

    # Step 4: If paid, fulfill the order
    if payment_status == 'paid':
        fulfill_order(order)
    else:
        # If payment fails, release inventory and cancel the order
        undo_reserve_inventory(order, items)
        cancel_order(order)

# Create a named queue for indexing tasks
queue = Queue("indexing_queue")

# Define a durable workflow that indexes a list of URLs
@DBOS.workflow()
def indexing_workflow(urls: List[HttpUrl]):
    handles: List[WorkflowHandle] = []

    # Enqueue a document indexing task for each URL
    for url in urls:
        handle = queue.enqueue(index_document, url)
        handles.append(handle)

    indexed_pages = 0

    # Wait for each indexing task to complete and tally results
    for handle in handles:
        indexed_pages += handle.get_result()

    # Log the total indexed pages
    logger.info(f"Indexed {len(urls)} documents totaling {indexed_pages} pages")

# Listen for new messages on the "alerts-topic" Kafka topic
@DBOS.kafka_consumer(config, ["alerts-topic"])

# Define a durable workflow triggered by each Kafka message
@DBOS.workflow()
def process_kafka_alerts(msg: KafkaMessage):
    # Decode the Kafka message payload
    alerts = msg.value.decode()
    
    # Loop through each alert and respond accordingly
    for alert in alerts:
        respond_to_alert(alert)

# Schedule this workflow to run every hour
@DBOS.scheduled("0 * * * *")

# Define a durable workflow that takes scheduled and actual time
@DBOS.workflow()
def run_hourly(scheduled_time: datetime, actual_time: datetime):
    # Search Hacker News for the keyword "serverless"
    results = search_hackernews("serverless")
    
    # Post each result (comment and URL) to Slack
    for comment, url in results:
        post_to_slack(comment, url)

@DBOS.workflow()
def schedule_reminder(to_email, days_to_wait):
    DBOS.recv(days_to_seconds(days_to_wait))
    send_reminder_email(to_email, days_to_wait)

@app.post("/email")
def email_endpoint(request):
    DBOS.start_workflow(
        schedule_reminder, 
        request.email, 
        request.days
)