# PostgreSQL Stored Procedure Reaction

A Drasi reaction plugin that invokes PostgreSQL stored procedures when continuous query results change.

## Overview

The PostgreSQL Stored Procedure reaction enables you to:
- Execute different stored procedures for ADD, UPDATE, and DELETE operations
- Build stored-procedure commands from [Handlebars](https://handlebarsjs.com/) templates, binding query-result fields as safe positional SQL parameters via the `{{param ...}}` helper
- Handle multiple queries with a single reaction
- Automatically retry failed procedure calls with exponential backoff
- Configure connection parameters and timeouts

## Installation

Add the dependency to your `Cargo.toml`:

```toml
[dependencies]
drasi-reaction-storedproc-postgres = { path = "path/to/drasi-core/components/reactions/storedproc-postgres" }
```

## Quick Start

### 1. Create Stored Procedures in PostgreSQL

```sql
CREATE OR REPLACE PROCEDURE add_user(
    p_id INTEGER,
    p_name TEXT,
    p_email TEXT
)
LANGUAGE plpgsql
AS $$
BEGIN
    INSERT INTO users_sync (id, name, email)
    VALUES (p_id, p_name, p_email);
END;
$$;
```

### 2. Create the Reaction

```rust
use drasi_reaction_storedproc_postgres::{PostgresStoredProcReaction, QueryConfig, TemplateSpec};
use drasi_lib::DrasiLib;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let reaction = PostgresStoredProcReaction::builder("user-sync")
        .with_connection(
            "localhost",
            5432,
            "mydb",
            "postgres",
            "password"
        )
        .with_query("user-changes")
        .with_default_template(QueryConfig {
            added: Some(TemplateSpec::new("CALL add_user({{param after.id}}, {{param after.name}}, {{param after.email}})")),
            updated: Some(TemplateSpec::new("CALL update_user({{param after.id}}, {{param after.name}}, {{param after.email}})")),
            deleted: Some(TemplateSpec::new("CALL delete_user({{param before.id}})")),
        })
        .build()
        .await?;

    let drasi = DrasiLib::builder()
        .with_id("my-app")
        .with_reaction(reaction)
        .build()
        .await?;

    drasi.start().await?;
    tokio::signal::ctrl_c().await?;

    Ok(())
}
```

## Configuration

### Builder API

#### Username/Password Authentication

```rust
let reaction = PostgresStoredProcReaction::builder("my-reaction")
    .with_hostname("localhost")
    .with_port(5432)
    .with_database("mydb")
    .with_user("postgres")
    .with_password("secret")
    .with_ssl(true)  // Enable SSL/TLS
    .with_query("query1")
    .with_default_template(QueryConfig {
        added: Some(TemplateSpec::new("CALL add_record({{param after.id}}, {{param after.name}})")),
        updated: Some(TemplateSpec::new("CALL update_record({{param after.id}}, {{param after.name}})")),
        deleted: Some(TemplateSpec::new("CALL delete_record({{param before.id}})")),
    })
    .with_command_timeout_ms(30000)
    .with_retry_attempts(3)
    .build()
    .await?;
```

#### Cloud Identity Provider Authentication

For cloud-managed PostgreSQL databases, you can use identity providers instead of passwords:

**Azure AD Authentication (Azure Database for PostgreSQL):**

```rust
use drasi_lib::identity::AzureIdentityProvider;

// For Azure Kubernetes Service with Workload Identity
let identity_provider = AzureIdentityProvider::with_workload_identity("myuser@myserver")?;

// For local development or Azure VMs with Managed Identity
let identity_provider = AzureIdentityProvider::with_default_credentials("myuser@myserver").await?;

let reaction = PostgresStoredProcReaction::builder("my-reaction")
    .with_hostname("myserver.postgres.database.azure.com")
    .with_port(5432)
    .with_database("mydb")
    .with_identity_provider(identity_provider)
    .with_ssl(true)  // Required for Azure
    .with_query("query1")
    .with_default_template(QueryConfig {
        added: Some(TemplateSpec::new("CALL add_record({{param after.id}}, {{param after.name}})")),
        updated: Some(TemplateSpec::new("CALL update_record({{param after.id}}, {{param after.name}})")),
        deleted: Some(TemplateSpec::new("CALL delete_record({{param before.id}})")),
    })
    .build()
    .await?;
```

**AWS IAM Authentication (Amazon RDS/Aurora):**

```rust
use drasi_identity_aws::AwsIdentityProvider;

// Using IAM user credentials
let identity_provider = AwsIdentityProvider::new(
    "myuser",
).await?;

// Or assuming an IAM role
let identity_provider = AwsIdentityProvider::with_assumed_role(
    "myuser",
    "arn:aws:iam::123456789012:role/RDSAccessRole",
    None
).await?;

let reaction = PostgresStoredProcReaction::builder("my-reaction")
    .with_hostname("mydb.rds.amazonaws.com")
    .with_port(5432)
    .with_database("mydb")
    .with_identity_provider(identity_provider)
    .with_ssl(true)  // Recommended for RDS
    .with_query("query1")
    .with_default_template(QueryConfig {
        added: Some(TemplateSpec::new("CALL add_record({{param after.id}}, {{param after.name}})")),
        updated: Some(TemplateSpec::new("CALL update_record({{param after.id}}, {{param after.name}})")),
        deleted: Some(TemplateSpec::new("CALL delete_record({{param before.id}})")),
    })
    .build()
    .await?;
```

**Password Provider (programmatic username/password):**

```rust
use drasi_lib::identity::PasswordIdentityProvider;

let identity_provider = PasswordIdentityProvider::new("postgres", "secret");

let reaction = PostgresStoredProcReaction::builder("my-reaction")
    .with_hostname("localhost")
    .with_port(5432)
    .with_database("mydb")
    .with_identity_provider(identity_provider)
    .with_query("query1")
    .with_default_template(QueryConfig {
        added: Some(TemplateSpec::new("CALL add_record({{param after.id}}, {{param after.name}})")),
        updated: Some(TemplateSpec::new("CALL update_record({{param after.id}}, {{param after.name}})")),
        deleted: Some(TemplateSpec::new("CALL delete_record({{param before.id}})")),
    })
    .build()
    .await?;
```

> **Note:** When using identity providers, do not call `.with_user()` or `.with_password()`. The identity provider handles authentication automatically.
>
> See the [Identity Provider README](../../../lib/src/identity/README.md) for detailed setup instructions for Azure AD and AWS IAM authentication.

### Configuration Options

| Option | Description | Type | Default |
|--------|-------------|------|---------|
| `hostname` | Database hostname | `String` | `"localhost"` |
| `port` | Database port | `u16` | `5432` |
| `user` | Database user | `String` | Required |
| `password` | Database password | `String` | Required |
| `database` | Database name | `String` | Required |
| `ssl` | Enable SSL/TLS | `bool` | `false` |
| `default_template` | Default templates for all queries | `Option<QueryConfig>` | `None` |
| `routes` | Query-specific template overrides | `HashMap<String, QueryConfig>` | Empty |
| `command_timeout_ms` | Command timeout | `u64` | `30000` |
| `retry_attempts` | Number of retries | `u32` | `3` |

## Command Templates

Stored-procedure commands are [Handlebars](https://handlebarsjs.com/) templates. Every
template is compiled when the reaction is built, so an invalid template fails fast at
construction time rather than at runtime.

### Template context

Each template is rendered against a standard context with these keys:

| Key | Description | Available for |
|-----|-------------|---------------|
| `after` | The post-change row | ADD, UPDATE |
| `before` | The pre-change row | UPDATE, DELETE |
| `data` | The raw `data` payload of an Update diff | UPDATE |
| `query_id` / `query_name` | The id of the query that produced the result (aliases holding the same value) | all |
| `operation` | `"ADD"`, `"UPDATE"`, or `"DELETE"` | all |
| `timestamp` | RFC3339 result timestamp | all |
| `metadata` | Result metadata map | all |

### Binding values with `{{param}}`

Reference a field with the `{{param <expr>}}` helper. Instead of inlining the value into
the SQL text, the helper appends the resolved value to the command's positional bind
parameters and emits a `$N` placeholder. Untrusted row data therefore can never alter the
command structure — it is always sent to PostgreSQL as a bound parameter.

For example, with this template:

```rust
TemplateSpec::new("CALL add_user({{param after.id}}, {{param after.name}}, {{param after.email}})")
```

and this ADD result row:

```json
{
  "id": 1,
  "name": "Alice",
  "email": "alice@example.com"
}
```

the reaction executes the SQL:

```sql
CALL add_user($1, $2, $3)
```

with `$1 = 1`, `$2 = "Alice"`, `$3 = "alice@example.com"` bound as parameters.

### Nested field access

Access nested fields using dot notation:

```rust
TemplateSpec::new("CALL add_address({{param after.user.id}}, {{param after.address.city}})")
```

### Binding a whole object as JSONB

To pass an entire object to a `jsonb` argument, reference it with `{{param}}` directly.
The object is bound as a single positional JSONB parameter, so it is safe even for
untrusted row data:

```rust
TemplateSpec::new("CALL ingest_record({{param after}})")
```

Only `{{param <expr>}}` may emit a value into the rendered SQL. Bare (`{{expr}}`) and raw
(`{{{expr}}}`) interpolation, as well as other helpers, are rejected when the reaction is
built so row data can never be inlined into SQL text. Write the procedure name and any
other structure as literal text in the template.

### Render failures and SQL safety

If a template references a missing field, or otherwise fails to render, the reaction logs
the error and **skips** that event rather than executing partial or unsafe SQL. If no
template is configured for an operation, the event is skipped.

## Template Resolution Order

For each `(query_id, operation)` the reaction resolves the command template in this order:

1. a `routes` entry keyed by the **full** query id,
2. a `routes` entry keyed by the query id's **last dotted segment** (so `source.my_query`
   can be routed via a `my_query` key),
3. the shared `default_template`.

If none of these supplies a template for the operation, the event is skipped. Every
`routes` key must match a subscribed query id (or its last dotted segment); an unmatched
key is rejected when the reaction is built.

### Per-Query Templates

Configure different templates for specific queries using the `routes` field or the
builder's `with_route` method:

```rust
let reaction = PostgresStoredProcReaction::builder("my-reaction")
    .with_hostname("localhost")
    .with_database("mydb")
    .with_user("postgres")
    .with_password("secret")
    .with_query("user-query")
    .with_route("user-query", QueryConfig {
        added: Some(TemplateSpec::new("CALL user_added({{param after.id}}, {{param after.name}})")),
        ..Default::default()  // updated and deleted will use default template
    })
    .build()
    .await?;
```

## Advanced Example: Partial Route Overrides

This example shows how to override only specific operations for a query while falling back to defaults for others:

```rust
use drasi_reaction_storedproc_postgres::{PostgresStoredProcReaction, QueryConfig, TemplateSpec};

let reaction = PostgresStoredProcReaction::builder("multi-query-sensor-sync")
    .with_hostname("localhost")
    .with_port(5432)
    .with_database("drasi_test")
    .with_user("postgres")
    .with_password("mysecret")
    // Subscribe to multiple queries
    .with_query("high-temp")
    .with_query("low-temp")
    .with_query("critical-temp")
    // Default template - applies to "high-temp" and "low-temp"
    .with_default_template(QueryConfig {
        added: Some(TemplateSpec::new(
            "CALL log_sensor_added({{param after.id}}, {{param after.temperature}}, {{param after.timestamp}})"
        )),
        updated: Some(TemplateSpec::new(
            "CALL log_sensor_updated({{param after.id}}, {{param after.temperature}})"
        )),
        deleted: Some(TemplateSpec::new(
            "CALL log_sensor_deleted({{param before.id}})"
        )),
    })
    // Custom route for critical temperature readings
    // Only handles ADD operations, falls back to default for UPDATE/DELETE
    .with_route("critical-temp", QueryConfig {
        added: Some(TemplateSpec::new(
            "CALL log_critical_alert({{param after.id}}, {{param after.temperature}}, {{param after.timestamp}})"
        )),
        ..Default::default()  // updated and deleted will use default template
    })
    .with_command_timeout_ms(5000)
    .with_retry_attempts(3)
    .build()
    .await?;
```

**How it works:**

- **"high-temp" and "low-temp" queries** → Use default template for all operations
- **"critical-temp" query**:
  - ADD: `CALL log_critical_alert(...)` *(custom route)*
  - UPDATE: `CALL log_sensor_updated(...)` *(falls back to default)*
  - DELETE: `CALL log_sensor_deleted(...)` *(falls back to default)*

## Advanced Example: Multiple Queries with Default and Custom Routes

This example shows a reaction handling multiple queries with different stored procedure requirements:

```rust
use drasi_reaction_storedproc_postgres::{PostgresStoredProcReaction, QueryConfig, TemplateSpec};

// Create a reaction that:
// 1. Subscribes to 3 different queries: "user-changes", "product-changes", "order-changes"
// 2. Has a default template for most operations
// 3. Overrides only the "product-changes" query with custom procedures

let reaction = PostgresStoredProcReaction::builder("multi-query-sync")
    .with_hostname("localhost")
    .with_port(5432)
    .with_database("mydb")
    .with_user("postgres")
    .with_password("secret")
    // Subscribe to multiple queries
    .with_query("user-changes")
    .with_query("product-changes")
    .with_query("order-changes")
    // Default template applies to "user-changes" and "order-changes"
    .with_default_template(QueryConfig {
        added: Some(TemplateSpec::new("CALL log_entity_added({{param after.id}}, {{param after.type}})")),
        updated: Some(TemplateSpec::new("CALL log_entity_updated({{param after.id}}, {{param after.type}})")),
        deleted: Some(TemplateSpec::new("CALL log_entity_deleted({{param before.id}}, {{param before.type}})")),
    })
    // Override "product-changes" with specific procedures
    .with_route("product-changes", QueryConfig {
        added: Some(TemplateSpec::new(
            "CALL sync_product_added({{param after.product_id}}, {{param after.name}}, {{param after.price}}, {{param after.inventory}})"
        )),
        updated: Some(TemplateSpec::new(
            "CALL sync_product_updated({{param after.product_id}}, {{param after.price}}, {{param after.inventory}})"
        )),
        ..Default::default()  // deleted will fall back to default template
    })
    .with_command_timeout_ms(5000)
    .with_retry_attempts(3)
    .build()
    .await?;
```

**How it works:**

1. **"user-changes" query** → Uses default template
   - Add: `CALL log_entity_added({{param after.id}}, {{param after.type}})`
   - Update: `CALL log_entity_updated({{param after.id}}, {{param after.type}})`
   - Delete: `CALL log_entity_deleted({{param before.id}}, {{param before.type}})`

2. **"product-changes" query** → Uses custom route (with fallback to default for delete)
   - Add: `CALL sync_product_added({{param after.product_id}}, {{param after.name}}, {{param after.price}}, {{param after.inventory}})`
   - Update: `CALL sync_product_updated({{param after.product_id}}, {{param after.price}}, {{param after.inventory}})`
   - Delete: `CALL log_entity_deleted({{param before.id}}, {{param before.type}})` *(falls back to default)*

3. **"order-changes" query** → Uses default template
   - Add: `CALL log_entity_added({{param after.id}}, {{param after.type}})`
   - Update: `CALL log_entity_updated({{param after.id}}, {{param after.type}})`
   - Delete: `CALL log_entity_deleted({{param before.id}}, {{param before.type}})`

**Note:** If a route specifies `None` for an operation (like `deleted: None` for product-changes), the reaction will check the default template. If the default template also has `None` for that operation, no procedure will be called.

## Error Handling

The reaction includes automatic retry logic with exponential backoff:
- **Initial retry**: 100ms delay
- **Subsequent retries**: 200ms, 400ms, 800ms, etc.
- **Max retries**: Configurable (default: 3)
- **Timeout**: Configurable per command (default: 30s)

## Plugin Packaging

This reaction is compiled as a dynamic plugin (cdylib) that can be loaded by drasi-server at runtime.

**Key files:**
- `Cargo.toml` — includes `crate-type = ["lib", "cdylib"]`
- `src/descriptor.rs` — implements `ReactionPluginDescriptor` with kind `"storedproc-postgres"`, configuration DTO, and OpenAPI schema generation
- `src/lib.rs` — invokes `drasi_plugin_sdk::export_plugin!` to export the plugin entry point

**Building:**
```bash
cargo build -p drasi-reaction-storedproc-postgres
```

The compiled `.so` (Linux) / `.dylib` (macOS) / `.dll` (Windows) is placed in `target/debug/` and can be copied to the server's `plugins/` directory.

For more details on the plugin descriptor pattern and configuration DTOs, see the [Reaction Developer Guide](../README.md#packaging-as-a-dynamic-plugin).

## License

Copyright 2025 The Drasi Authors.

Licensed under the Apache License, Version 2.0.
