view connectors/store.c @ 279:b3b547563ec7

Add Google connector service and agent wiki Implement the C/Seobeo Google Drive and Gmail connector with encrypted OAuth storage, Zenbu authentication, browser testing, AI tool discovery, chunked HTTP decoding, and Bazel coverage. Consolidate repository guidance into progressive wiki documentation and enforce arena-first allocation for new first-party C code. Co-authored-by: Copilot <[email protected]> Copilot-Session: 84c338fd-0939-4bb3-b7f3-1062eb213e5d
author MrJuneJune <me@mrjunejune.com>
date Mon, 17 Aug 2026 22:22:36 -0700
parents
children
line wrap: on
line source

#include "connectors/connector.h"

#include <stdio.h>
#include <string.h>

static boolean execute(Connector_Store *store, const char *sql)
{
  return store && store->connection &&
      Deita_Query_Execute_Update(store->connection, sql) >= 0;
}

static boolean update(
    Connector_Store *store, const char *sql, int32 count, const char **values)
{
  return store && store->connection &&
      Deita_Query_Execute_Update_Prepared(
          store->connection, sql, count, values) >= 0;
}

boolean Connector_Store_Open(
    Connector_Store *store, const char *database_path,
    const Connector_Master_Key *master_key)
{
  if (!store || !database_path || !master_key)
    return FALSE;
  memset(store, 0, sizeof(*store));
  store->connection = Deita_Connection_Create(
      DEITA_DATABASE_TYPE_SQLITE3, database_path);
  if (!store->connection)
    return FALSE;
  store->master_key = *master_key;
  return Connector_Store_Migrate(store);
}

void Connector_Store_Close(Connector_Store *store)
{
  if (!store)
    return;
  if (store->connection)
    Deita_Connection_Close(store->connection);
  memset(store, 0, sizeof(*store));
}

boolean Connector_Store_Migrate(Connector_Store *store)
{
  static const char *migration =
      "BEGIN;"
      "CREATE TABLE IF NOT EXISTS connector_migrations("
      "version INTEGER PRIMARY KEY, applied_at INTEGER NOT NULL);"
      "CREATE TABLE IF NOT EXISTS connector_oauth_states("
      "state TEXT PRIMARY KEY, user_id TEXT NOT NULL, verifier TEXT NOT NULL,"
      "expires_at INTEGER NOT NULL, consumed_at INTEGER);"
      "CREATE TABLE IF NOT EXISTS connector_accounts("
      "account_id TEXT PRIMARY KEY, user_id TEXT NOT NULL, provider TEXT NOT NULL,"
      "provider_subject TEXT NOT NULL, email TEXT NOT NULL,"
      "access_token_enc TEXT NOT NULL, refresh_token_enc TEXT NOT NULL,"
      "expires_at INTEGER NOT NULL, scopes TEXT NOT NULL,"
      "UNIQUE(user_id, provider, provider_subject));"
      "CREATE INDEX IF NOT EXISTS connector_accounts_owner "
      "ON connector_accounts(user_id, account_id);"
      "CREATE TABLE IF NOT EXISTS connector_policies("
      "user_id TEXT NOT NULL, action TEXT NOT NULL, policy INTEGER NOT NULL,"
      "PRIMARY KEY(user_id, action));"
      "CREATE TABLE IF NOT EXISTS connector_confirmations("
      "token TEXT PRIMARY KEY, user_id TEXT NOT NULL, digest TEXT NOT NULL,"
      "expires_at INTEGER NOT NULL, consumed_at INTEGER);"
      "CREATE TABLE IF NOT EXISTS connector_idempotency("
      "user_id TEXT NOT NULL, idem_key TEXT NOT NULL, digest TEXT NOT NULL,"
      "status INTEGER NOT NULL, response_body TEXT NOT NULL,"
      "created_at INTEGER NOT NULL DEFAULT (unixepoch()),"
      "PRIMARY KEY(user_id, idem_key));"
      "CREATE TABLE IF NOT EXISTS connector_audit("
      "id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL,"
      "account_id TEXT NOT NULL, operation TEXT NOT NULL, digest TEXT NOT NULL,"
      "status INTEGER NOT NULL, created_at INTEGER NOT NULL DEFAULT (unixepoch()));"
      "INSERT OR IGNORE INTO connector_migrations(version, applied_at)"
      "VALUES(1, unixepoch());"
      "COMMIT;";
  return execute(store, "PRAGMA foreign_keys=ON;") &&
      execute(store, migration);
}

boolean Connector_Store_Create_State(
    Connector_Store *store, const char *user_id, const Connector_OAuth_Start *start,
    int64 expires_at)
{
  char expiry[32];
  snprintf(expiry, sizeof(expiry), "%lld", expires_at);
  const char *values[] = {
    start ? start->state : NULL, user_id,
    start ? start->code_verifier : NULL, expiry
  };
  return start && user_id && update(
      store,
      "INSERT INTO connector_oauth_states"
      "(state,user_id,verifier,expires_at) VALUES(?,?,?,?)",
      4, values);
}

boolean Connector_Store_Consume_State(
    Connector_Store *store, const char *user_id, const char *state, int64 now,
    char *verifier, size_t verifier_size)
{
  if (!store || !user_id || !state || !verifier || verifier_size == 0)
    return FALSE;
  char now_text[32];
  snprintf(now_text, sizeof(now_text), "%lld", now);
  const char *values[] = {state, user_id, now_text};
  Dowa_Arena *arena = Dowa_Arena_Create(4096);
  Deita_Result_Set *result = Deita_Query_Execute_Prepared(
      store->connection,
      "SELECT verifier FROM connector_oauth_states "
      "WHERE state=? AND user_id=? AND consumed_at IS NULL AND expires_at>=?",
      3, values, arena);
  boolean found = result && Deita_Result_Set_Next(result);
  if (found) {
    const char *value = Deita_Result_Set_Get_Text(result, 0);
    if (!value || strlen(value) >= verifier_size)
      found = FALSE;
    else
      strcpy(verifier, value);
  }
  if (result)
    Deita_Result_Set_Free(result);
  Dowa_Arena_Free(arena);
  if (!found)
    return FALSE;
  const char *consume_values[] = {now_text, state, user_id};
  return Deita_Query_Execute_Update_Prepared(
      store->connection,
      "UPDATE connector_oauth_states SET consumed_at=? "
      "WHERE state=? AND user_id=? AND consumed_at IS NULL",
      3, consume_values) == 1;
}

boolean Connector_Store_Save_Account(
    Connector_Store *store, const Connector_Account *account)
{
  if (!store || !account || !account->account_id[0] || !account->user_id[0])
    return FALSE;
  char access[4096], refresh[4096], expires[32];
  if (!Connector_Encrypt(&store->master_key, account->access_token, access,
                         sizeof(access)) ||
      !Connector_Encrypt(&store->master_key, account->refresh_token, refresh,
                         sizeof(refresh)))
    return FALSE;
  snprintf(expires, sizeof(expires), "%lld", account->expires_at);
  const char *values[] = {
    account->account_id, account->user_id, account->provider,
    account->provider_subject, account->email, access, refresh, expires,
    account->scopes
  };
  return update(
      store,
      "INSERT INTO connector_accounts(account_id,user_id,provider,"
      "provider_subject,email,access_token_enc,refresh_token_enc,expires_at,scopes)"
      "VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(account_id) DO UPDATE SET "
      "email=excluded.email,access_token_enc=excluded.access_token_enc,"
      "refresh_token_enc=excluded.refresh_token_enc,"
      "expires_at=excluded.expires_at,scopes=excluded.scopes "
      "WHERE connector_accounts.user_id=excluded.user_id",
      9, values);
}

boolean Connector_Store_Get_Account(
    Connector_Store *store, const char *user_id, const char *account_id,
    Connector_Account *account)
{
  if (!store || !user_id || !account_id || !account)
    return FALSE;
  const char *values[] = {user_id, account_id};
  Dowa_Arena *arena = Dowa_Arena_Create(16384);
  Deita_Result_Set *result = Deita_Query_Execute_Prepared(
      store->connection,
      "SELECT account_id,user_id,provider,provider_subject,email,"
      "access_token_enc,refresh_token_enc,expires_at,scopes "
      "FROM connector_accounts WHERE user_id=? AND account_id=?",
      2, values, arena);
  boolean found = result && Deita_Result_Set_Next(result);
  if (found) {
    memset(account, 0, sizeof(*account));
#define COPY_COLUMN(field, index) do { \
  const char *value = Deita_Result_Set_Get_Text(result, index); \
  if (!value || strlen(value) >= sizeof(account->field)) found = FALSE; \
  else strcpy(account->field, value); \
} while (0)
    COPY_COLUMN(account_id, 0);
    COPY_COLUMN(user_id, 1);
    COPY_COLUMN(provider, 2);
    COPY_COLUMN(provider_subject, 3);
    COPY_COLUMN(email, 4);
    const char *access = Deita_Result_Set_Get_Text(result, 5);
    const char *refresh = Deita_Result_Set_Get_Text(result, 6);
    account->expires_at = Deita_Result_Set_Get_Integer(result, 7);
    COPY_COLUMN(scopes, 8);
    if (!access || !refresh ||
        !Connector_Decrypt(&store->master_key, access, account->access_token,
                           sizeof(account->access_token)) ||
        !Connector_Decrypt(&store->master_key, refresh, account->refresh_token,
                           sizeof(account->refresh_token)))
      found = FALSE;
#undef COPY_COLUMN
  }
  if (result)
    Deita_Result_Set_Free(result);
  Dowa_Arena_Free(arena);
  return found;
}

boolean Connector_Store_List_Accounts(
    Connector_Store *store, const char *user_id,
    Connector_Account_Summary **accounts, Dowa_Arena *arena)
{
  if (!store || !user_id || !accounts || !arena)
    return FALSE;
  *accounts = NULL;
  const char *values[] = {user_id};
  Deita_Result_Set *result = Deita_Query_Execute_Prepared(
      store->connection,
      "SELECT account_id,provider,email,expires_at,scopes "
      "FROM connector_accounts WHERE user_id=? ORDER BY provider,email",
      1, values, arena);
  if (!result || Deita_Result_Set_Has_Error(result)) {
    if (result)
      Deita_Result_Set_Free(result);
    return FALSE;
  }
  while (Deita_Result_Set_Next(result)) {
    Connector_Account_Summary account = {0};
    const char *account_id = Deita_Result_Set_Get_Text(result, 0);
    const char *provider = Deita_Result_Set_Get_Text(result, 1);
    const char *email = Deita_Result_Set_Get_Text(result, 2);
    const char *scopes = Deita_Result_Set_Get_Text(result, 4);
    if (!account_id || !provider || !email || !scopes ||
        strlen(account_id) >= sizeof(account.account_id) ||
        strlen(provider) >= sizeof(account.provider) ||
        strlen(email) >= sizeof(account.email) ||
        strlen(scopes) >= sizeof(account.scopes)) {
      Deita_Result_Set_Free(result);
      return FALSE;
    }
    strcpy(account.account_id, account_id);
    strcpy(account.provider, provider);
    strcpy(account.email, email);
    account.expires_at = Deita_Result_Set_Get_Integer(result, 3);
    strcpy(account.scopes, scopes);
    Dowa_Array_Push_Arena(*accounts, account, arena);
  }
  Deita_Result_Set_Free(result);
  return TRUE;
}

boolean Connector_Store_Delete_Account(
    Connector_Store *store, const char *user_id, const char *account_id)
{
  const char *values[] = {user_id, account_id};
  return user_id && account_id &&
      Deita_Query_Execute_Update_Prepared(
          store->connection,
          "DELETE FROM connector_accounts WHERE user_id=? AND account_id=?",
          2, values) == 1;
}

boolean Connector_Store_Set_Policy(
    Connector_Store *store, const char *user_id, const char *action,
    Connector_Confirmation_Policy policy)
{
  char policy_text[8];
  snprintf(policy_text, sizeof(policy_text), "%d", policy);
  const char *values[] = {user_id, action, policy_text};
  return user_id && action && update(
      store,
      "INSERT INTO connector_policies(user_id,action,policy) VALUES(?,?,?) "
      "ON CONFLICT(user_id,action) DO UPDATE SET policy=excluded.policy",
      3, values);
}

Connector_Confirmation_Policy Connector_Store_Get_Policy(
    Connector_Store *store, const char *user_id, const char *action)
{
  Connector_Confirmation_Policy fallback =
      action && (!strcmp(action, "gmail.send") ||
                 !strcmp(action, "drive.overwrite"))
          ? CONNECTOR_CONFIRM_ALWAYS : CONNECTOR_CONFIRM_NEVER;
  if (!store || !user_id || !action)
    return fallback;
  const char *values[] = {user_id, action};
  Dowa_Arena *arena = Dowa_Arena_Create(2048);
  Deita_Result_Set *result = Deita_Query_Execute_Prepared(
      store->connection,
      "SELECT policy FROM connector_policies WHERE user_id=? AND action=?",
      2, values, arena);
  if (result && Deita_Result_Set_Next(result))
    fallback = (Connector_Confirmation_Policy)
        Deita_Result_Set_Get_Integer(result, 0);
  if (result)
    Deita_Result_Set_Free(result);
  Dowa_Arena_Free(arena);
  return fallback;
}

boolean Connector_Store_Create_Confirmation(
    Connector_Store *store, const char *user_id, const char *digest,
    const char *token, int64 expires_at)
{
  char expiry[32];
  snprintf(expiry, sizeof(expiry), "%lld", expires_at);
  const char *values[] = {token, user_id, digest, expiry};
  return token && user_id && digest && update(
      store,
      "INSERT INTO connector_confirmations(token,user_id,digest,expires_at)"
      "VALUES(?,?,?,?)", 4, values);
}

boolean Connector_Store_Consume_Confirmation(
    Connector_Store *store, const char *user_id, const char *digest,
    const char *token, int64 now)
{
  char now_text[32];
  snprintf(now_text, sizeof(now_text), "%lld", now);
  const char *values[] = {now_text, token, user_id, digest, now_text};
  return token && user_id && digest &&
      Deita_Query_Execute_Update_Prepared(
          store->connection,
          "UPDATE connector_confirmations SET consumed_at=? WHERE token=? "
          "AND user_id=? AND digest=? AND consumed_at IS NULL AND expires_at>=?",
          5, values) == 1;
}

boolean Connector_Store_Get_Idempotent(
    Connector_Store *store, const char *user_id, const char *key,
    const char *digest, int32 *status, char *body, size_t body_size)
{
  if (!store || !user_id || !key || !digest || !status || !body)
    return FALSE;
  const char *values[] = {user_id, key, digest};
  Dowa_Arena *arena = Dowa_Arena_Create(4096 + body_size);
  Deita_Result_Set *result = Deita_Query_Execute_Prepared(
      store->connection,
      "SELECT status,response_body FROM connector_idempotency "
      "WHERE user_id=? AND idem_key=? AND digest=?",
      3, values, arena);
  boolean found = result && Deita_Result_Set_Next(result);
  if (found) {
    const char *value = Deita_Result_Set_Get_Text(result, 1);
    if (!value || strlen(value) >= body_size)
      found = FALSE;
    else {
      *status = (int32)Deita_Result_Set_Get_Integer(result, 0);
      strcpy(body, value);
    }
  }
  if (result)
    Deita_Result_Set_Free(result);
  Dowa_Arena_Free(arena);
  return found;
}

boolean Connector_Store_Put_Idempotent(
    Connector_Store *store, const char *user_id, const char *key,
    const char *digest, int32 status, const char *body)
{
  char status_text[16];
  snprintf(status_text, sizeof(status_text), "%d", status);
  const char *values[] = {user_id, key, digest, status_text, body ? body : ""};
  return user_id && key && digest && update(
      store,
      "INSERT OR IGNORE INTO connector_idempotency"
      "(user_id,idem_key,digest,status,response_body) VALUES(?,?,?,?,?)",
      5, values);
}

boolean Connector_Store_Idempotency_Conflict(
    Connector_Store *store, const char *user_id, const char *key,
    const char *digest)
{
  if (!store || !user_id || !key || !digest)
    return FALSE;
  const char *values[] = {user_id, key, digest};
  Dowa_Arena *arena = Dowa_Arena_Create(2048);
  Deita_Result_Set *result = Deita_Query_Execute_Prepared(
      store->connection,
      "SELECT 1 FROM connector_idempotency "
      "WHERE user_id=? AND idem_key=? AND digest<>?",
      3, values, arena);
  boolean conflict = result && Deita_Result_Set_Next(result);
  if (result)
    Deita_Result_Set_Free(result);
  Dowa_Arena_Free(arena);
  return conflict;
}

boolean Connector_Store_Audit(
    Connector_Store *store, const char *user_id, const char *account_id,
    const char *operation, const char *digest, int32 status)
{
  char status_text[16];
  snprintf(status_text, sizeof(status_text), "%d", status);
  const char *values[] = {
    user_id, account_id, operation, digest, status_text
  };
  return user_id && account_id && operation && digest && update(
      store,
      "INSERT INTO connector_audit"
      "(user_id,account_id,operation,digest,status) VALUES(?,?,?,?,?)",
      5, values);
}