apps/whatsapp_control_api/app/services/store.py
text
from __future__ import annotations
import json
from datetime import datetime, timedelta, timezone
from uuid import uuid4
from ..core.config import get_settings
from ..db import dumps_json, get_connection
from .task_registry import sync_store_task_to_registry
def utcnow() -> datetime:
return datetime.now(timezone.utc)
def utcnow_iso() -> str:
return utcnow().isoformat()
class Store:
def get_or_create_user(self, whatsapp_number: str, display_name: str | None = None) -> dict:
with get_connection() as conn:
row = conn.execute(
"SELECT * FROM users WHERE whatsapp_number = ?",
(whatsapp_number,),
).fetchone()
if row:
conn.execute(
"UPDATE users SET display_name = COALESCE(?, display_name), last_seen_at = ?, updated_at = ? WHERE id = ?",
(display_name, utcnow_iso(), utcnow_iso(), row["id"]),
)
conn.commit()
return dict(row)
user_id = uuid4().hex
now = utcnow_iso()
conn.execute(
"""
INSERT INTO users (id, whatsapp_number, display_name, role, status, created_at, updated_at, last_seen_at)
VALUES (?, ?, ?, 'owner', 'active', ?, ?, ?)
""",
(user_id, whatsapp_number, display_name, now, now, now),
)
conn.commit()
return {
"id": user_id,
"whatsapp_number": whatsapp_number,
"display_name": display_name,
"role": "owner",
"status": "active",
}
def get_user_by_number(self, whatsapp_number: str) -> dict | None:
with get_connection() as conn:
row = conn.execute(
"SELECT * FROM users WHERE whatsapp_number = ?",
(whatsapp_number,),
).fetchone()
return dict(row) if row else None
def get_or_create_session(self, user_id: str, channel_chat_id: str, *, channel: str = "whatsapp") -> dict:
with get_connection() as conn:
row = conn.execute(
"SELECT * FROM sessions WHERE user_id = ? AND channel = ? AND channel_chat_id = ?",
(user_id, channel, channel_chat_id),
).fetchone()
now = utcnow_iso()
if row:
conn.execute(
"UPDATE sessions SET last_message_at = ?, updated_at = ? WHERE id = ?",
(now, now, row["id"]),
)
conn.commit()
return dict(row)
session_id = uuid4().hex
conn.execute(
"""
INSERT INTO sessions (id, user_id, channel, channel_chat_id, state, last_message_at, context_summary, created_at, updated_at)
VALUES (?, ?, ?, ?, 'active', ?, NULL, ?, ?)
""",
(session_id, user_id, channel, channel_chat_id, now, now, now),
)
conn.commit()
return {
"id": session_id,
"user_id": user_id,
"channel": channel,
"channel_chat_id": channel_chat_id,
"state": "active",
}
def webhook_event_exists(self, delivery_key: str) -> bool:
with get_connection() as conn:
row = conn.execute(
"SELECT 1 FROM webhook_events WHERE delivery_key = ?",
(delivery_key,),
).fetchone()
return row is not None
def create_webhook_event(self, delivery_key: str, payload: dict, signature_valid: bool, event_type: str) -> None:
with get_connection() as conn:
conn.execute(
"""
INSERT INTO webhook_events (id, provider, event_type, delivery_key, signature_valid, payload_json, received_at)
VALUES (?, 'meta_whatsapp', ?, ?, ?, ?, ?)
""",
(uuid4().hex, event_type, delivery_key, int(signature_valid), dumps_json(payload), utcnow_iso()),
)
conn.commit()
def create_message(
self,
*,
session_id: str | None,
user_id: str | None,
direction: str,
message_type: str,
wa_message_id: str | None,
reply_to_wa_id: str | None,
body_text: str | None,
payload: dict,
status: str,
) -> str:
message_id = uuid4().hex
with get_connection() as conn:
conn.execute(
"""
INSERT INTO messages (id, session_id, user_id, direction, message_type, wa_message_id, reply_to_wa_id, body_text, payload_json, status, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
message_id,
session_id,
user_id,
direction,
message_type,
wa_message_id,
reply_to_wa_id,
body_text,
dumps_json(payload),
status,
utcnow_iso(),
),
)
conn.commit()
return message_id
def create_task(
self,
*,
user_id: str,
session_id: str,
source_message_id: str,
prompt: str,
status: str,
channel: str = "whatsapp",
) -> str:
task_id = uuid4().hex[:12]
now = utcnow_iso()
with get_connection() as conn:
conn.execute(
"""
INSERT INTO tasks (id, user_id, session_id, source_message_id, project_name, channel, prompt, status, current_step, progress_percent, created_at, updated_at)
VALUES (?, ?, ?, ?, NULL, ?, ?, ?, 'Queued', 0, ?, ?)
""",
(task_id, user_id, session_id, source_message_id, channel, prompt, status, now, now),
)
conn.commit()
self._sync_registry(task_id)
return task_id
def get_task(self, task_id: str) -> dict | None:
with get_connection() as conn:
row = conn.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
return dict(row) if row else None
def get_recent_tasks_for_user(self, user_id: str, *, limit: int = 5) -> list[dict]:
with get_connection() as conn:
rows = conn.execute(
"""
SELECT * FROM tasks
WHERE user_id = ?
ORDER BY created_at DESC
LIMIT ?
""",
(user_id, limit),
).fetchall()
return [dict(row) for row in rows]
def get_recent_tasks(self, *, limit: int = 12) -> list[dict]:
with get_connection() as conn:
rows = conn.execute(
"""
SELECT * FROM tasks
ORDER BY created_at DESC
LIMIT ?
""",
(limit,),
).fetchall()
return [dict(row) for row in rows]
def get_latest_task_for_user(self, user_id: str) -> dict | None:
with get_connection() as conn:
row = conn.execute(
"""
SELECT * FROM tasks
WHERE user_id = ?
ORDER BY created_at DESC
LIMIT 1
""",
(user_id,),
).fetchone()
return dict(row) if row else None
def get_task_logs(self, task_id: str, *, limit: int = 100) -> list[dict]:
with get_connection() as conn:
rows = conn.execute(
"""
SELECT * FROM task_logs
WHERE task_id = ?
ORDER BY id DESC
LIMIT ?
""",
(task_id, limit),
).fetchall()
return [dict(row) for row in reversed(rows)]
def update_task(self, task_id: str, **fields: object) -> None:
if not fields:
return
fields["updated_at"] = utcnow_iso()
assignments = ", ".join(f"{key} = ?" for key in fields)
values = list(fields.values()) + [task_id]
with get_connection() as conn:
conn.execute(f"UPDATE tasks SET {assignments} WHERE id = ?", values)
conn.commit()
self._sync_registry(task_id)
def append_task_log(self, task_id: str, level: str, message: str) -> None:
with get_connection() as conn:
conn.execute(
"INSERT INTO task_logs (task_id, level, message, created_at) VALUES (?, ?, ?, ?)",
(task_id, level, message, utcnow_iso()),
)
conn.commit()
self._sync_registry(task_id)
def _sync_registry(self, task_id: str) -> None:
try:
task = self.get_task(task_id)
if not task:
return
logs = self.get_task_logs(task_id, limit=24)
sync_store_task_to_registry(task, logs, get_settings())
except Exception:
return
def create_approval(
self,
*,
task_id: str,
requested_by_user_id: str,
approval_type: str,
reason: str,
risk_summary: str,
approval_code: str,
nonce_hash: str,
ttl_seconds: int,
) -> str:
approval_id = uuid4().hex
requested_at = utcnow()
expires_at = requested_at + timedelta(seconds=ttl_seconds)
with get_connection() as conn:
conn.execute(
"""
INSERT INTO approvals (
id, task_id, requested_by_user_id, approved_by_user_id, approval_type, reason, risk_summary,
status, approval_code, nonce_hash, requested_at, responded_at, expires_at
) VALUES (?, ?, ?, NULL, ?, ?, ?, 'pending', ?, ?, ?, NULL, ?)
""",
(
approval_id,
task_id,
requested_by_user_id,
approval_type,
reason,
risk_summary,
approval_code,
nonce_hash,
requested_at.isoformat(),
expires_at.isoformat(),
),
)
conn.commit()
return approval_id
def get_pending_approval(self, task_id: str) -> dict | None:
with get_connection() as conn:
row = conn.execute(
"""
SELECT * FROM approvals
WHERE task_id = ? AND status = 'pending'
ORDER BY requested_at DESC LIMIT 1
""",
(task_id,),
).fetchone()
return dict(row) if row else None
def update_approval(self, approval_id: str, **fields: object) -> None:
if not fields:
return
assignments = ", ".join(f"{key} = ?" for key in fields)
values = list(fields.values()) + [approval_id]
with get_connection() as conn:
conn.execute(f"UPDATE approvals SET {assignments} WHERE id = ?", values)
conn.commit()
def create_audit_log(
self,
*,
actor_type: str,
actor_ref: str,
action: str,
target_type: str,
target_ref: str,
metadata: dict,
) -> None:
with get_connection() as conn:
conn.execute(
"""
INSERT INTO audit_logs (actor_type, actor_ref, action, target_type, target_ref, metadata_json, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(actor_type, actor_ref, action, target_type, target_ref, json.dumps(metadata, ensure_ascii=True), utcnow_iso()),
)
conn.commit()
def create_config_change(
self,
*,
change_id: str,
user_id: str,
scope: str,
target_name: str | None,
request_json: dict,
request_summary: str,
reason: str,
status: str = "pending",
) -> None:
with get_connection() as conn:
conn.execute(
"""
INSERT INTO config_changes (
id, user_id, scope, target_name, status, request_json, request_summary, reason,
requested_at, responded_at, applied_at, error_summary
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL)
""",
(
change_id,
user_id,
scope,
target_name,
status,
json.dumps(request_json, ensure_ascii=True),
request_summary,
reason,
utcnow_iso(),
),
)
conn.commit()
def get_config_change(self, change_id: str) -> dict | None:
with get_connection() as conn:
row = conn.execute("SELECT * FROM config_changes WHERE id = ?", (change_id,)).fetchone()
return dict(row) if row else None
def update_config_change(self, change_id: str, **fields: object) -> None:
if not fields:
return
assignments = ", ".join(f"{key} = ?" for key in fields)
values = list(fields.values()) + [change_id]
with get_connection() as conn:
conn.execute(f"UPDATE config_changes SET {assignments} WHERE id = ?", values)
conn.commit()