feat: 0.6.0 — 事务/版本化/类型化/inject/run

- 写路径事务化：add/update/delete 与 audit 同事务，update CAS 并发保护
- 版本化与回滚：secrets_history 表、version 字段、history/rollback 命令
- 类型化字段：key:=<json> 支持数字、布尔、数组、对象
- 临时 env 模式：inject 输出 KEY=VALUE，run 向子进程注入
- inject/run 至少需一个过滤条件；search -o env 使用 shell_quote；JSON 输出含 version

Made-with: Cursor

src/commands/add.rs

@@ -4,13 +4,30 @@ use sqlx::PgPool;
 use std::fs;
 
 use crate::crypto;
+use crate::db;
 use crate::output::OutputMode;
 
-/// Parse "key=value" entries. Value starting with '@' reads from file.
-pub(crate) fn parse_kv(entry: &str) -> Result<(String, String)> {
+/// Parse "key=value" or "key:=<json>" entries.
+/// - `key=value`   → stores the literal string `value`
+/// - `key:=<json>` → parses `<json>` as a typed JSON value (number, bool, null, array, object)
+/// - `value=@file` → reads the file content as a string (only for `=` form)
+pub(crate) fn parse_kv(entry: &str) -> Result<(String, Value)> {
+    // Typed JSON form: key:=<json>
+    if let Some((key, json_str)) = entry.split_once(":=") {
+        let val: Value = serde_json::from_str(json_str).map_err(|e| {
+            anyhow::anyhow!(
+                "Invalid JSON value for key '{}': {} (use key=value for plain strings)",
+                key,
+                e
+            )
+        })?;
+        return Ok((key.to_string(), val));
+    }
+
+    // Plain string form: key=value or key=@file
     let (key, raw_val) = entry.split_once('=').ok_or_else(|| {
         anyhow::anyhow!(
-            "Invalid format '{}'. Expected: key=value or key=@file",
+            "Invalid format '{}'. Expected: key=value, key=@file, or key:=<json>",
             entry
         )
     })?;
@@ -22,14 +39,14 @@ pub(crate) fn parse_kv(entry: &str) -> Result<(String, String)> {
         raw_val.to_string()
     };
 
-    Ok((key.to_string(), value))
+    Ok((key.to_string(), Value::String(value)))
 }
 
 pub(crate) fn build_json(entries: &[String]) -> Result<Value> {
     let mut map = Map::new();
     for entry in entries {
         let (key, value) = parse_kv(entry)?;
-        map.insert(key, Value::String(value));
+        map.insert(key, value);
     }
     Ok(Value::Object(map))
 }
@@ -47,21 +64,72 @@ pub struct AddArgs<'a> {
 pub async fn run(pool: &PgPool, args: AddArgs<'_>, master_key: &[u8; 32]) -> Result<()> {
     let metadata = build_json(args.meta_entries)?;
     let secret_json = build_json(args.secret_entries)?;
-
-    // Encrypt the secret JSON before storing
     let encrypted_bytes = crypto::encrypt_json(master_key, &secret_json)?;
 
     tracing::debug!(args.namespace, args.kind, args.name, "upserting record");
 
+    let meta_keys: Vec<&str> = args
+        .meta_entries
+        .iter()
+        .filter_map(|s| s.split_once(['=', ':']).map(|(k, _)| k))
+        .collect();
+    let secret_keys: Vec<&str> = args
+        .secret_entries
+        .iter()
+        .filter_map(|s| s.split_once(['=', ':']).map(|(k, _)| k))
+        .collect();
+
+    let mut tx = pool.begin().await?;
+
+    // Snapshot existing row into history before overwriting (if it exists).
+    #[derive(sqlx::FromRow)]
+    struct ExistingRow {
+        id: uuid::Uuid,
+        version: i64,
+        tags: Vec<String>,
+        metadata: serde_json::Value,
+        encrypted: Vec<u8>,
+    }
+    let existing: Option<ExistingRow> = sqlx::query_as(
+        "SELECT id, version, tags, metadata, encrypted FROM secrets \
+         WHERE namespace = $1 AND kind = $2 AND name = $3",
+    )
+    .bind(args.namespace)
+    .bind(args.kind)
+    .bind(args.name)
+    .fetch_optional(&mut *tx)
+    .await?;
+
+    if let Some(ex) = existing
+        && let Err(e) = db::snapshot_history(
+            &mut tx,
+            db::SnapshotParams {
+                secret_id: ex.id,
+                namespace: args.namespace,
+                kind: args.kind,
+                name: args.name,
+                version: ex.version,
+                action: "add",
+                tags: &ex.tags,
+                metadata: &ex.metadata,
+                encrypted: &ex.encrypted,
+            },
+        )
+        .await
+    {
+        tracing::warn!(error = %e, "failed to snapshot history before upsert");
+    }
+
     sqlx::query(
         r#"
-        INSERT INTO secrets (namespace, kind, name, tags, metadata, encrypted, updated_at)
-        VALUES ($1, $2, $3, $4, $5, $6, NOW())
+        INSERT INTO secrets (namespace, kind, name, tags, metadata, encrypted, version, updated_at)
+        VALUES ($1, $2, $3, $4, $5, $6, 1, NOW())
         ON CONFLICT (namespace, kind, name)
         DO UPDATE SET
-            tags      = EXCLUDED.tags,
-            metadata  = EXCLUDED.metadata,
-            encrypted = EXCLUDED.encrypted,
+            tags       = EXCLUDED.tags,
+            metadata   = EXCLUDED.metadata,
+            encrypted  = EXCLUDED.encrypted,
+            version    = secrets.version + 1,
             updated_at = NOW()
         "#,
     )
@@ -71,22 +139,11 @@ pub async fn run(pool: &PgPool, args: AddArgs<'_>, master_key: &[u8; 32]) -> Res
     .bind(args.tags)
     .bind(&metadata)
     .bind(&encrypted_bytes)
-    .execute(pool)
+    .execute(&mut *tx)
     .await?;
 
-    let meta_keys: Vec<&str> = args
-        .meta_entries
-        .iter()
-        .filter_map(|s| s.split_once('=').map(|(k, _)| k))
-        .collect();
-    let secret_keys: Vec<&str> = args
-        .secret_entries
-        .iter()
-        .filter_map(|s| s.split_once('=').map(|(k, _)| k))
-        .collect();
-
-    crate::audit::log(
-        pool,
+    crate::audit::log_tx(
+        &mut tx,
         "add",
         args.namespace,
         args.kind,
@@ -99,6 +156,8 @@ pub async fn run(pool: &PgPool, args: AddArgs<'_>, master_key: &[u8; 32]) -> Res
     )
     .await;
 
+    tx.commit().await?;
+
     let result_json = json!({
         "action": "added",
         "namespace": args.namespace,

src/commands/delete.rs

@@ -1,9 +1,20 @@
 use anyhow::Result;
-use serde_json::json;
-use sqlx::PgPool;
+use serde_json::{Value, json};
+use sqlx::{FromRow, PgPool};
+use uuid::Uuid;
 
+use crate::db;
 use crate::output::OutputMode;
 
+#[derive(FromRow)]
+struct DeleteRow {
+    id: Uuid,
+    version: i64,
+    tags: Vec<String>,
+    metadata: Value,
+    encrypted: Vec<u8>,
+}
+
 pub async fn run(
     pool: &PgPool,
     namespace: &str,
@@ -13,15 +24,21 @@ pub async fn run(
 ) -> Result<()> {
     tracing::debug!(namespace, kind, name, "deleting record");
 
-    let result =
-        sqlx::query("DELETE FROM secrets WHERE namespace = $1 AND kind = $2 AND name = $3")
-            .bind(namespace)
-            .bind(kind)
-            .bind(name)
-            .execute(pool)
-            .await?;
+    let mut tx = pool.begin().await?;
 
-    if result.rows_affected() == 0 {
+    let row: Option<DeleteRow> = sqlx::query_as(
+        "SELECT id, version, tags, metadata, encrypted FROM secrets \
+         WHERE namespace = $1 AND kind = $2 AND name = $3 \
+         FOR UPDATE",
+    )
+    .bind(namespace)
+    .bind(kind)
+    .bind(name)
+    .fetch_optional(&mut *tx)
+    .await?;
+
+    let Some(row) = row else {
+        tx.rollback().await?;
         tracing::warn!(namespace, kind, name, "record not found for deletion");
         match output {
             OutputMode::Json => println!(
@@ -38,23 +55,53 @@ pub async fn run(
             ),
             _ => println!("Not found: [{}/{}] {}", namespace, kind, name),
         }
-    } else {
-        crate::audit::log(pool, "delete", namespace, kind, name, json!({})).await;
-        match output {
-            OutputMode::Json => println!(
-                "{}",
-                serde_json::to_string_pretty(
-                    &json!({"action":"deleted","namespace":namespace,"kind":kind,"name":name})
-                )?
-            ),
-            OutputMode::JsonCompact => println!(
-                "{}",
-                serde_json::to_string(
-                    &json!({"action":"deleted","namespace":namespace,"kind":kind,"name":name})
-                )?
-            ),
-            _ => println!("Deleted: [{}/{}] {}", namespace, kind, name),
-        }
+        return Ok(());
+    };
+
+    // Snapshot before physical delete so the row can be restored via rollback.
+    if let Err(e) = db::snapshot_history(
+        &mut tx,
+        db::SnapshotParams {
+            secret_id: row.id,
+            namespace,
+            kind,
+            name,
+            version: row.version,
+            action: "delete",
+            tags: &row.tags,
+            metadata: &row.metadata,
+            encrypted: &row.encrypted,
+        },
+    )
+    .await
+    {
+        tracing::warn!(error = %e, "failed to snapshot history before delete");
     }
+
+    sqlx::query("DELETE FROM secrets WHERE id = $1")
+        .bind(row.id)
+        .execute(&mut *tx)
+        .await?;
+
+    crate::audit::log_tx(&mut tx, "delete", namespace, kind, name, json!({})).await;
+
+    tx.commit().await?;
+
+    match output {
+        OutputMode::Json => println!(
+            "{}",
+            serde_json::to_string_pretty(
+                &json!({"action":"deleted","namespace":namespace,"kind":kind,"name":name})
+            )?
+        ),
+        OutputMode::JsonCompact => println!(
+            "{}",
+            serde_json::to_string(
+                &json!({"action":"deleted","namespace":namespace,"kind":kind,"name":name})
+            )?
+        ),
+        _ => println!("Deleted: [{}/{}] {}", namespace, kind, name),
+    }
+
     Ok(())
 }

src/commands/mod.rs

@@ -2,5 +2,7 @@ pub mod add;
 pub mod config;
 pub mod delete;
 pub mod init;
+pub mod rollback;
+pub mod run;
 pub mod search;
 pub mod update;

src/commands/rollback.rs

@@ -0,0 +1,245 @@
+use anyhow::Result;
+use serde_json::{Value, json};
+use sqlx::{FromRow, PgPool};
+use uuid::Uuid;
+
+use crate::output::OutputMode;
+
+#[derive(FromRow)]
+struct HistoryRow {
+    secret_id: Uuid,
+    #[allow(dead_code)]
+    namespace: String,
+    #[allow(dead_code)]
+    kind: String,
+    #[allow(dead_code)]
+    name: String,
+    version: i64,
+    action: String,
+    tags: Vec<String>,
+    metadata: Value,
+    encrypted: Vec<u8>,
+}
+
+pub struct RollbackArgs<'a> {
+    pub namespace: &'a str,
+    pub kind: &'a str,
+    pub name: &'a str,
+    /// Target version to restore. None → restore the most recent history entry.
+    pub to_version: Option<i64>,
+    pub output: OutputMode,
+}
+
+pub async fn run(pool: &PgPool, args: RollbackArgs<'_>, master_key: &[u8; 32]) -> Result<()> {
+    let snap: Option<HistoryRow> = if let Some(ver) = args.to_version {
+        sqlx::query_as(
+            "SELECT secret_id, namespace, kind, name, version, action, tags, metadata, encrypted \
+             FROM secrets_history \
+             WHERE namespace = $1 AND kind = $2 AND name = $3 AND version = $4 \
+             ORDER BY id DESC LIMIT 1",
+        )
+        .bind(args.namespace)
+        .bind(args.kind)
+        .bind(args.name)
+        .bind(ver)
+        .fetch_optional(pool)
+        .await?
+    } else {
+        sqlx::query_as(
+            "SELECT secret_id, namespace, kind, name, version, action, tags, metadata, encrypted \
+             FROM secrets_history \
+             WHERE namespace = $1 AND kind = $2 AND name = $3 \
+             ORDER BY id DESC LIMIT 1",
+        )
+        .bind(args.namespace)
+        .bind(args.kind)
+        .bind(args.name)
+        .fetch_optional(pool)
+        .await?
+    };
+
+    let snap = snap.ok_or_else(|| {
+        anyhow::anyhow!(
+            "No history found for [{}/{}] {}{}.",
+            args.namespace,
+            args.kind,
+            args.name,
+            args.to_version
+                .map(|v| format!(" at version {}", v))
+                .unwrap_or_default()
+        )
+    })?;
+
+    // Validate encrypted blob is non-trivial (re-encrypt guard).
+    if !snap.encrypted.is_empty() {
+        // Probe decrypt to ensure the blob is valid before restoring.
+        crate::crypto::decrypt_json(master_key, &snap.encrypted)?;
+    }
+
+    let mut tx = pool.begin().await?;
+
+    // Snapshot current live row (if it exists) before overwriting.
+    #[derive(sqlx::FromRow)]
+    struct LiveRow {
+        id: Uuid,
+        version: i64,
+        tags: Vec<String>,
+        metadata: Value,
+        encrypted: Vec<u8>,
+    }
+    let live: Option<LiveRow> = sqlx::query_as(
+        "SELECT id, version, tags, metadata, encrypted FROM secrets \
+         WHERE namespace = $1 AND kind = $2 AND name = $3 FOR UPDATE",
+    )
+    .bind(args.namespace)
+    .bind(args.kind)
+    .bind(args.name)
+    .fetch_optional(&mut *tx)
+    .await?;
+
+    if let Some(lr) = live
+        && let Err(e) = crate::db::snapshot_history(
+            &mut tx,
+            crate::db::SnapshotParams {
+                secret_id: lr.id,
+                namespace: args.namespace,
+                kind: args.kind,
+                name: args.name,
+                version: lr.version,
+                action: "rollback",
+                tags: &lr.tags,
+                metadata: &lr.metadata,
+                encrypted: &lr.encrypted,
+            },
+        )
+        .await
+    {
+        tracing::warn!(error = %e, "failed to snapshot current row before rollback");
+    }
+
+    sqlx::query(
+        "INSERT INTO secrets (id, namespace, kind, name, tags, metadata, encrypted, version, updated_at) \
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW()) \
+         ON CONFLICT (namespace, kind, name) DO UPDATE SET \
+             tags       = EXCLUDED.tags, \
+             metadata   = EXCLUDED.metadata, \
+             encrypted  = EXCLUDED.encrypted, \
+             version    = secrets.version + 1, \
+             updated_at = NOW()",
+    )
+    .bind(snap.secret_id)
+    .bind(args.namespace)
+    .bind(args.kind)
+    .bind(args.name)
+    .bind(&snap.tags)
+    .bind(&snap.metadata)
+    .bind(&snap.encrypted)
+    .bind(snap.version)
+    .execute(&mut *tx)
+    .await?;
+
+    crate::audit::log_tx(
+        &mut tx,
+        "rollback",
+        args.namespace,
+        args.kind,
+        args.name,
+        json!({
+            "restored_version": snap.version,
+            "original_action": snap.action,
+        }),
+    )
+    .await;
+
+    tx.commit().await?;
+
+    let result_json = json!({
+        "action": "rolled_back",
+        "namespace": args.namespace,
+        "kind": args.kind,
+        "name": args.name,
+        "restored_version": snap.version,
+    });
+
+    match args.output {
+        OutputMode::Json => println!("{}", serde_json::to_string_pretty(&result_json)?),
+        OutputMode::JsonCompact => println!("{}", serde_json::to_string(&result_json)?),
+        _ => println!(
+            "Rolled back: [{}/{}] {} → version {}",
+            args.namespace, args.kind, args.name, snap.version
+        ),
+    }
+
+    Ok(())
+}
+
+/// List history entries for a record.
+pub async fn list_history(
+    pool: &PgPool,
+    namespace: &str,
+    kind: &str,
+    name: &str,
+    limit: u32,
+    output: OutputMode,
+) -> Result<()> {
+    #[derive(FromRow)]
+    struct HistorySummary {
+        version: i64,
+        action: String,
+        actor: String,
+        created_at: chrono::DateTime<chrono::Utc>,
+    }
+
+    let rows: Vec<HistorySummary> = sqlx::query_as(
+        "SELECT version, action, actor, created_at FROM secrets_history \
+         WHERE namespace = $1 AND kind = $2 AND name = $3 \
+         ORDER BY id DESC LIMIT $4",
+    )
+    .bind(namespace)
+    .bind(kind)
+    .bind(name)
+    .bind(limit as i64)
+    .fetch_all(pool)
+    .await?;
+
+    match output {
+        OutputMode::Json | OutputMode::JsonCompact => {
+            let arr: Vec<Value> = rows
+                .iter()
+                .map(|r| {
+                    json!({
+                        "version": r.version,
+                        "action": r.action,
+                        "actor": r.actor,
+                        "created_at": r.created_at.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
+                    })
+                })
+                .collect();
+            let out = if output == OutputMode::Json {
+                serde_json::to_string_pretty(&arr)?
+            } else {
+                serde_json::to_string(&arr)?
+            };
+            println!("{}", out);
+        }
+        _ => {
+            if rows.is_empty() {
+                println!("No history found for [{}/{}] {}.", namespace, kind, name);
+                return Ok(());
+            }
+            println!("History for [{}/{}] {}:", namespace, kind, name);
+            for r in &rows {
+                println!(
+                    "  v{:<4}  {:8}  {}  {}",
+                    r.version,
+                    r.action,
+                    r.actor,
+                    r.created_at.format("%Y-%m-%d %H:%M:%S UTC")
+                );
+            }
+            println!("  (use `secrets rollback --to-version <N>` to restore)");
+        }
+    }
+
+    Ok(())
+}

src/commands/run.rs

@@ -0,0 +1,143 @@
+use anyhow::Result;
+use serde_json::Value;
+use sqlx::PgPool;
+use std::collections::HashMap;
+
+use crate::commands::search::build_env_map;
+use crate::output::OutputMode;
+
+pub struct InjectArgs<'a> {
+    pub namespace: Option<&'a str>,
+    pub kind: Option<&'a str>,
+    pub name: Option<&'a str>,
+    pub tags: &'a [String],
+    /// Prefix to prepend to every variable name. Empty string means no prefix.
+    pub prefix: &'a str,
+    pub output: OutputMode,
+}
+
+pub struct RunArgs<'a> {
+    pub namespace: Option<&'a str>,
+    pub kind: Option<&'a str>,
+    pub name: Option<&'a str>,
+    pub tags: &'a [String],
+    pub prefix: &'a str,
+    /// The command and its arguments to execute with injected secrets.
+    pub command: &'a [String],
+}
+
+/// Fetch secrets matching the filter and build a flat env map.
+/// Metadata and secret fields are merged; naming: `<PREFIX_><NAME>_<KEY>` (uppercased).
+pub async fn collect_env_map(
+    pool: &PgPool,
+    namespace: Option<&str>,
+    kind: Option<&str>,
+    name: Option<&str>,
+    tags: &[String],
+    prefix: &str,
+    master_key: &[u8; 32],
+) -> Result<HashMap<String, String>> {
+    if namespace.is_none() && kind.is_none() && name.is_none() && tags.is_empty() {
+        anyhow::bail!(
+            "At least one filter (--namespace, --kind, --name, or --tag) is required for inject/run"
+        );
+    }
+    let rows = crate::commands::search::fetch_rows(pool, namespace, kind, name, tags, None).await?;
+    if rows.is_empty() {
+        anyhow::bail!("No records matched the given filters.");
+    }
+    let mut map = HashMap::new();
+    for row in &rows {
+        let row_map = build_env_map(row, prefix, Some(master_key))?;
+        for (k, v) in row_map {
+            map.insert(k, v);
+        }
+    }
+    Ok(map)
+}
+
+/// `inject` command: print env vars to stdout (suitable for `eval $(...)` or export).
+pub async fn run_inject(pool: &PgPool, args: InjectArgs<'_>, master_key: &[u8; 32]) -> Result<()> {
+    let env_map = collect_env_map(
+        pool,
+        args.namespace,
+        args.kind,
+        args.name,
+        args.tags,
+        args.prefix,
+        master_key,
+    )
+    .await?;
+
+    match args.output {
+        OutputMode::Json => {
+            let obj: serde_json::Map<String, Value> = env_map
+                .into_iter()
+                .map(|(k, v)| (k, Value::String(v)))
+                .collect();
+            println!("{}", serde_json::to_string_pretty(&Value::Object(obj))?);
+        }
+        OutputMode::JsonCompact => {
+            let obj: serde_json::Map<String, Value> = env_map
+                .into_iter()
+                .map(|(k, v)| (k, Value::String(v)))
+                .collect();
+            println!("{}", serde_json::to_string(&Value::Object(obj))?);
+        }
+        _ => {
+            // Shell-safe KEY=VALUE output, one per line.
+            let mut pairs: Vec<(String, String)> = env_map.into_iter().collect();
+            pairs.sort_by(|a, b| a.0.cmp(&b.0));
+            for (k, v) in pairs {
+                println!("{}={}", k, shell_quote(&v));
+            }
+        }
+    }
+
+    Ok(())
+}
+
+/// `run` command: inject secrets into a child process environment and execute.
+pub async fn run_exec(pool: &PgPool, args: RunArgs<'_>, master_key: &[u8; 32]) -> Result<()> {
+    if args.command.is_empty() {
+        anyhow::bail!(
+            "No command specified. Usage: secrets run [filter flags] -- <command> [args]"
+        );
+    }
+
+    let env_map = collect_env_map(
+        pool,
+        args.namespace,
+        args.kind,
+        args.name,
+        args.tags,
+        args.prefix,
+        master_key,
+    )
+    .await?;
+
+    tracing::debug!(
+        vars = env_map.len(),
+        cmd = args.command[0].as_str(),
+        "injecting secrets into child process"
+    );
+
+    let status = std::process::Command::new(&args.command[0])
+        .args(&args.command[1..])
+        .envs(&env_map)
+        .status()
+        .map_err(|e| anyhow::anyhow!("Failed to execute '{}': {}", args.command[0], e))?;
+
+    if !status.success() {
+        let code = status.code().unwrap_or(1);
+        std::process::exit(code);
+    }
+
+    Ok(())
+}
+
+/// Quote a value for safe shell output. Wraps the value in single quotes,
+/// escaping any single quotes within the value.
+fn shell_quote(s: &str) -> String {
+    format!("'{}'", s.replace('\'', "'\\''"))
+}

src/commands/search.rs

@@ -1,6 +1,7 @@
 use anyhow::Result;
 use serde_json::{Value, json};
 use sqlx::PgPool;
+use std::collections::HashMap;
 
 use crate::crypto;
 use crate::models::Secret;
@@ -22,88 +23,20 @@ pub struct SearchArgs<'a> {
 }
 
 pub async fn run(pool: &PgPool, args: SearchArgs<'_>, master_key: Option<&[u8; 32]>) -> Result<()> {
-    let mut conditions: Vec<String> = Vec::new();
-    let mut idx: i32 = 1;
-
-    if args.namespace.is_some() {
-        conditions.push(format!("namespace = ${}", idx));
-        idx += 1;
-    }
-    if args.kind.is_some() {
-        conditions.push(format!("kind = ${}", idx));
-        idx += 1;
-    }
-    if args.name.is_some() {
-        conditions.push(format!("name = ${}", idx));
-        idx += 1;
-    }
-    if !args.tags.is_empty() {
-        // Use PostgreSQL array containment: tags @> ARRAY[$n, $m, ...] means all specified tags must be present
-        let placeholders: Vec<String> = args
-            .tags
-            .iter()
-            .map(|_| {
-                let p = format!("${}", idx);
-                idx += 1;
-                p
-            })
-            .collect();
-        conditions.push(format!("tags @> ARRAY[{}]", placeholders.join(", ")));
-    }
-    if args.query.is_some() {
-        conditions.push(format!(
-            "(name ILIKE ${i} ESCAPE '\\' OR namespace ILIKE ${i} ESCAPE '\\' OR kind ILIKE ${i} ESCAPE '\\' OR metadata::text ILIKE ${i} ESCAPE '\\' OR EXISTS (SELECT 1 FROM unnest(tags) t WHERE t ILIKE ${i} ESCAPE '\\'))",
-            i = idx
-        ));
-        idx += 1;
-    }
-
-    let where_clause = if conditions.is_empty() {
-        String::new()
-    } else {
-        format!("WHERE {}", conditions.join(" AND "))
-    };
-
-    let order = match args.sort {
-        "updated" => "updated_at DESC",
-        "created" => "created_at DESC",
-        _ => "namespace, kind, name",
-    };
-
-    let sql = format!(
-        "SELECT * FROM secrets {} ORDER BY {} LIMIT ${} OFFSET ${}",
-        where_clause,
-        order,
-        idx,
-        idx + 1
-    );
-
-    tracing::debug!(sql, "executing search query");
-
-    let mut q = sqlx::query_as::<_, Secret>(&sql);
-    if let Some(v) = args.namespace {
-        q = q.bind(v);
-    }
-    if let Some(v) = args.kind {
-        q = q.bind(v);
-    }
-    if let Some(v) = args.name {
-        q = q.bind(v);
-    }
-    for v in args.tags {
-        q = q.bind(v.as_str());
-    }
-    if let Some(v) = args.query {
-        q = q.bind(format!(
-            "%{}%",
-            v.replace('\\', "\\\\")
-                .replace('%', "\\%")
-                .replace('_', "\\_")
-        ));
-    }
-    q = q.bind(args.limit as i64).bind(args.offset as i64);
-
-    let rows = q.fetch_all(pool).await?;
+    let rows = fetch_rows_paged(
+        pool,
+        PagedFetchArgs {
+            namespace: args.namespace,
+            kind: args.kind,
+            name: args.name,
+            tags: args.tags,
+            query: args.query,
+            sort: args.sort,
+            limit: args.limit,
+            offset: args.offset,
+        },
+    )
+    .await?;
 
     // -f/--field: extract specific field values directly
     if !args.fields.is_empty() {
@@ -131,7 +64,12 @@ pub async fn run(pool: &PgPool, args: SearchArgs<'_>, master_key: Option<&[u8; 3
                 );
             }
             if let Some(row) = rows.first() {
-                print_env(row, args.show_secrets, master_key)?;
+                let map = build_env_map(row, "", master_key)?;
+                let mut pairs: Vec<(String, String)> = map.into_iter().collect();
+                pairs.sort_by(|a, b| a.0.cmp(&b.0));
+                for (k, v) in pairs {
+                    println!("{}={}", k, shell_quote(&v));
+                }
             } else {
                 eprintln!("No records found.");
             }
@@ -158,8 +96,195 @@ pub async fn run(pool: &PgPool, args: SearchArgs<'_>, master_key: Option<&[u8; 3
     Ok(())
 }
 
+/// Fetch rows with simple equality/tag filters (no pagination). Used by inject/run.
+pub async fn fetch_rows(
+    pool: &PgPool,
+    namespace: Option<&str>,
+    kind: Option<&str>,
+    name: Option<&str>,
+    tags: &[String],
+    query: Option<&str>,
+) -> Result<Vec<Secret>> {
+    fetch_rows_paged(
+        pool,
+        PagedFetchArgs {
+            namespace,
+            kind,
+            name,
+            tags,
+            query,
+            sort: "name",
+            limit: 200,
+            offset: 0,
+        },
+    )
+    .await
+}
+
+/// Arguments for the internal paged fetch. Grouped to avoid too-many-arguments lint.
+struct PagedFetchArgs<'a> {
+    namespace: Option<&'a str>,
+    kind: Option<&'a str>,
+    name: Option<&'a str>,
+    tags: &'a [String],
+    query: Option<&'a str>,
+    sort: &'a str,
+    limit: u32,
+    offset: u32,
+}
+
+async fn fetch_rows_paged(pool: &PgPool, a: PagedFetchArgs<'_>) -> Result<Vec<Secret>> {
+    let mut conditions: Vec<String> = Vec::new();
+    let mut idx: i32 = 1;
+
+    if a.namespace.is_some() {
+        conditions.push(format!("namespace = ${}", idx));
+        idx += 1;
+    }
+    if a.kind.is_some() {
+        conditions.push(format!("kind = ${}", idx));
+        idx += 1;
+    }
+    if a.name.is_some() {
+        conditions.push(format!("name = ${}", idx));
+        idx += 1;
+    }
+    if !a.tags.is_empty() {
+        let placeholders: Vec<String> = a
+            .tags
+            .iter()
+            .map(|_| {
+                let p = format!("${}", idx);
+                idx += 1;
+                p
+            })
+            .collect();
+        conditions.push(format!("tags @> ARRAY[{}]", placeholders.join(", ")));
+    }
+    if a.query.is_some() {
+        conditions.push(format!(
+            "(name ILIKE ${i} ESCAPE '\\' OR namespace ILIKE ${i} ESCAPE '\\' OR kind ILIKE ${i} ESCAPE '\\' OR metadata::text ILIKE ${i} ESCAPE '\\' OR EXISTS (SELECT 1 FROM unnest(tags) t WHERE t ILIKE ${i} ESCAPE '\\'))",
+            i = idx
+        ));
+        idx += 1;
+    }
+
+    let where_clause = if conditions.is_empty() {
+        String::new()
+    } else {
+        format!("WHERE {}", conditions.join(" AND "))
+    };
+
+    let order = match a.sort {
+        "updated" => "updated_at DESC",
+        "created" => "created_at DESC",
+        _ => "namespace, kind, name",
+    };
+
+    let sql = format!(
+        "SELECT * FROM secrets {} ORDER BY {} LIMIT ${} OFFSET ${}",
+        where_clause,
+        order,
+        idx,
+        idx + 1
+    );
+
+    tracing::debug!(sql, "executing search query");
+
+    let mut q = sqlx::query_as::<_, Secret>(&sql);
+    if let Some(v) = a.namespace {
+        q = q.bind(v);
+    }
+    if let Some(v) = a.kind {
+        q = q.bind(v);
+    }
+    if let Some(v) = a.name {
+        q = q.bind(v);
+    }
+    for v in a.tags {
+        q = q.bind(v.as_str());
+    }
+    if let Some(v) = a.query {
+        q = q.bind(format!(
+            "%{}%",
+            v.replace('\\', "\\\\")
+                .replace('%', "\\%")
+                .replace('_', "\\_")
+        ));
+    }
+    q = q.bind(a.limit as i64).bind(a.offset as i64);
+
+    let rows = q.fetch_all(pool).await?;
+    Ok(rows)
+}
+
+/// Build a flat `KEY=VALUE` map from a record's metadata and decrypted secrets.
+/// Variable names: `<PREFIX><NAME>_<FIELD>` (all uppercased, hyphens/dots → underscores).
+/// If `prefix` is empty, the name segment alone is used as the prefix.
+pub fn build_env_map(
+    row: &Secret,
+    prefix: &str,
+    master_key: Option<&[u8; 32]>,
+) -> Result<HashMap<String, String>> {
+    let name_part = row.name.to_uppercase().replace(['-', '.', ' '], "_");
+    let effective_prefix = if prefix.is_empty() {
+        name_part
+    } else {
+        format!(
+            "{}_{}",
+            prefix.to_uppercase().replace(['-', '.', ' '], "_"),
+            name_part
+        )
+    };
+
+    let mut map = HashMap::new();
+
+    if let Some(meta) = row.metadata.as_object() {
+        for (k, v) in meta {
+            let key = format!(
+                "{}_{}",
+                effective_prefix,
+                k.to_uppercase().replace(['-', '.'], "_")
+            );
+            map.insert(key, json_value_to_env_string(v));
+        }
+    }
+
+    if let Some(master_key) = master_key
+        && !row.encrypted.is_empty()
+    {
+        let decrypted = crypto::decrypt_json(master_key, &row.encrypted)?;
+        if let Some(enc) = decrypted.as_object() {
+            for (k, v) in enc {
+                let key = format!(
+                    "{}_{}",
+                    effective_prefix,
+                    k.to_uppercase().replace(['-', '.'], "_")
+                );
+                map.insert(key, json_value_to_env_string(v));
+            }
+        }
+    }
+
+    Ok(map)
+}
+
+/// Quote a value for safe shell / env output. Wraps in single quotes,
+/// escaping any single quotes within the value.
+fn shell_quote(s: &str) -> String {
+    format!("'{}'", s.replace('\'', "'\\''"))
+}
+
+/// Convert a JSON value to its string representation suitable for env vars.
+fn json_value_to_env_string(v: &Value) -> String {
+    match v {
+        Value::String(s) => s.clone(),
+        Value::Null => String::new(),
+        other => other.to_string(),
+    }
+}
+
 /// Decrypt the encrypted blob for a row. Returns an empty object on empty blobs.
-/// Returns an error value on decrypt failure (so callers can decide how to handle).
 fn try_decrypt(row: &Secret, master_key: Option<&[u8; 32]>) -> Result<Value> {
     if row.encrypted.is_empty() {
         return Ok(Value::Object(Default::default()));
@@ -211,10 +336,12 @@ fn to_json(
         "tags": row.tags,
         "metadata": row.metadata,
         "secrets": secrets_val,
+        "version": row.version,
         "created_at": row.created_at.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
         "updated_at": row.updated_at.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
     })
 }
+
 fn print_text(
     row: &Secret,
     show_secrets: bool,
@@ -267,30 +394,9 @@ fn print_text(
     Ok(())
 }
 
-fn print_env(row: &Secret, show_secrets: bool, master_key: Option<&[u8; 32]>) -> Result<()> {
-    let prefix = row.name.to_uppercase().replace(['-', '.'], "_");
-    if let Some(meta) = row.metadata.as_object() {
-        for (k, v) in meta {
-            let key = format!("{}_{}", prefix, k.to_uppercase().replace('-', "_"));
-            println!("{}={}", key, v.as_str().unwrap_or(&v.to_string()));
-        }
-    }
-    if show_secrets {
-        let decrypted = try_decrypt(row, master_key)?;
-        if let Some(enc) = decrypted.as_object() {
-            for (k, v) in enc {
-                let key = format!("{}_{}", prefix, k.to_uppercase().replace('-', "_"));
-                println!("{}={}", key, v.as_str().unwrap_or(&v.to_string()));
-            }
-        }
-    }
-    Ok(())
-}
-
 /// Extract one or more field paths like `metadata.url` or `secret.token`.
 fn print_fields(rows: &[Secret], fields: &[String], master_key: Option<&[u8; 32]>) -> Result<()> {
     for row in rows {
-        // Decrypt once per row if any field requires it
         let decrypted: Option<Value> = if fields
             .iter()
             .any(|f| f.starts_with("secret") || f.starts_with("encrypted"))

src/commands/update.rs

@@ -5,11 +5,13 @@ use uuid::Uuid;
 
 use super::add::parse_kv;
 use crate::crypto;
+use crate::db;
 use crate::output::OutputMode;
 
 #[derive(FromRow)]
 struct UpdateRow {
     id: Uuid,
+    version: i64,
     tags: Vec<String>,
     metadata: Value,
     encrypted: Vec<u8>,
@@ -29,17 +31,18 @@ pub struct UpdateArgs<'a> {
 }
 
 pub async fn run(pool: &PgPool, args: UpdateArgs<'_>, master_key: &[u8; 32]) -> Result<()> {
+    let mut tx = pool.begin().await?;
+
     let row: Option<UpdateRow> = sqlx::query_as(
-        r#"
-        SELECT id, tags, metadata, encrypted
-        FROM secrets
-        WHERE namespace = $1 AND kind = $2 AND name = $3
-        "#,
+        "SELECT id, version, tags, metadata, encrypted \
+         FROM secrets \
+         WHERE namespace = $1 AND kind = $2 AND name = $3 \
+         FOR UPDATE",
     )
     .bind(args.namespace)
     .bind(args.kind)
     .bind(args.name)
-    .fetch_optional(pool)
+    .fetch_optional(&mut *tx)
     .await?;
 
     let row = row.ok_or_else(|| {
@@ -51,6 +54,26 @@ pub async fn run(pool: &PgPool, args: UpdateArgs<'_>, master_key: &[u8; 32]) ->
         )
     })?;
 
+    // Snapshot current state before modifying.
+    if let Err(e) = db::snapshot_history(
+        &mut tx,
+        db::SnapshotParams {
+            secret_id: row.id,
+            namespace: args.namespace,
+            kind: args.kind,
+            name: args.name,
+            version: row.version,
+            action: "update",
+            tags: &row.tags,
+            metadata: &row.metadata,
+            encrypted: &row.encrypted,
+        },
+    )
+    .await
+    {
+        tracing::warn!(error = %e, "failed to snapshot history before update");
+    }
+
     // Merge tags
     let mut tags: Vec<String> = row.tags;
     for t in args.add_tags {
@@ -67,7 +90,7 @@ pub async fn run(pool: &PgPool, args: UpdateArgs<'_>, master_key: &[u8; 32]) ->
     };
     for entry in args.meta_entries {
         let (key, value) = parse_kv(entry)?;
-        meta_map.insert(key, Value::String(value));
+        meta_map.insert(key, value);
     }
     for key in args.remove_meta {
         meta_map.remove(key);
@@ -86,7 +109,7 @@ pub async fn run(pool: &PgPool, args: UpdateArgs<'_>, master_key: &[u8; 32]) ->
     };
     for entry in args.secret_entries {
         let (key, value) = parse_kv(entry)?;
-        enc_map.insert(key, Value::String(value));
+        enc_map.insert(key, value);
     }
     for key in args.remove_secrets {
         enc_map.remove(key);
@@ -101,33 +124,43 @@ pub async fn run(pool: &PgPool, args: UpdateArgs<'_>, master_key: &[u8; 32]) ->
         "updating record"
     );
 
-    sqlx::query(
-        r#"
-        UPDATE secrets
-        SET tags = $1, metadata = $2, encrypted = $3, updated_at = NOW()
-        WHERE id = $4
-        "#,
+    // CAS: update only if version hasn't changed (FOR UPDATE lock ensures this).
+    let result = sqlx::query(
+        "UPDATE secrets \
+         SET tags = $1, metadata = $2, encrypted = $3, version = version + 1, updated_at = NOW() \
+         WHERE id = $4 AND version = $5",
     )
     .bind(&tags)
-    .bind(metadata)
-    .bind(encrypted_bytes)
+    .bind(&metadata)
+    .bind(&encrypted_bytes)
     .bind(row.id)
-    .execute(pool)
+    .bind(row.version)
+    .execute(&mut *tx)
     .await?;
 
+    if result.rows_affected() == 0 {
+        tx.rollback().await?;
+        anyhow::bail!(
+            "Concurrent modification detected for [{}/{}] {}. Please retry.",
+            args.namespace,
+            args.kind,
+            args.name
+        );
+    }
+
     let meta_keys: Vec<&str> = args
         .meta_entries
         .iter()
-        .filter_map(|s| s.split_once('=').map(|(k, _)| k))
+        .filter_map(|s| s.split_once(['=', ':']).map(|(k, _)| k))
         .collect();
     let secret_keys: Vec<&str> = args
         .secret_entries
         .iter()
-        .filter_map(|s| s.split_once('=').map(|(k, _)| k))
+        .filter_map(|s| s.split_once(['=', ':']).map(|(k, _)| k))
         .collect();
 
-    crate::audit::log(
-        pool,
+    crate::audit::log_tx(
+        &mut tx,
         "update",
         args.namespace,
         args.kind,
@@ -143,6 +176,8 @@ pub async fn run(pool: &PgPool, args: UpdateArgs<'_>, master_key: &[u8; 32]) ->
     )
     .await;
 
+    tx.commit().await?;
+
     let result_json = json!({
         "action": "updated",
         "namespace": args.namespace,