update-server/app/repositories/crash_report_repository.py

from datetime import datetime, timezone
import secrets

import db


def list_crash_reports(limit: int = 300) -> list[dict]:
    conn = db.get_conn()
    rows = conn.execute(
        """
        SELECT
          id,report_id,client_report_id,status,symbolication_status,product,app_version,
          git_commit,build_type,channel,exception_code,crash_time_utc,received_at_utc,
          remote_address,content_length,metadata_size,minidump_size,attachments_size
        FROM crash_reports
        ORDER BY id DESC
        LIMIT ?
        """,
        (limit,),
    ).fetchall()
    conn.close()
    return [dict(row) for row in rows]


def get_crash_report(report_id: str):
    conn = db.get_conn()
    row = conn.execute("SELECT * FROM crash_reports WHERE report_id=?", (report_id,)).fetchone()
    conn.close()
    return row


def log_file_access(report_id: str, file_name: str, actor_hash: str, ip: str, user_agent: str):
    conn = db.get_conn()
    conn.execute(
        "INSERT INTO crash_file_access_logs(report_id,file_name,actor_hash,ip,user_agent) VALUES(?,?,?,?,?)",
        (report_id, file_name, actor_hash, ip, user_agent),
    )
    conn.commit()
    conn.close()


def generate_prefixed_id(conn, table: str, column: str, prefix: str) -> str | None:
    day = datetime.now(timezone.utc).strftime("%Y%m%d")
    for _ in range(20):
        value = f"{prefix}_{day}_{secrets.token_hex(4)}"
        if not conn.execute(f"SELECT 1 FROM {table} WHERE {column}=?", (value,)).fetchone():
            return value
    return None


def get_report_by_client_report_id(conn, client_report_id: str):
    return conn.execute("SELECT * FROM crash_reports WHERE client_report_id=?", (client_report_id,)).fetchone()


def get_report_by_id(conn, report_id: str):
    return conn.execute("SELECT * FROM crash_reports WHERE report_id=?", (report_id,)).fetchone()


def insert_crash_report(conn, values: dict):
    conn.execute(
        """
        INSERT INTO crash_reports(
          report_id,client_report_id,content_hash,status,symbolication_status,product,app_version,
          git_commit,build_type,channel,exception_code,crash_time_utc,received_at_utc,remote_address,
          content_length,metadata_sha256,metadata_size,minidump_sha256,minidump_size,attachments_sha256,
          attachments_size,storage_path,metadata_json,server_json
        ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
        """,
        (
            values["report_id"],
            values["client_report_id"],
            values["content_hash"],
            values["status"],
            values["symbolication_status"],
            values["product"],
            values["app_version"],
            values["git_commit"],
            values["build_type"],
            values["channel"],
            values["exception_code"],
            values["crash_time_utc"],
            values["received_at_utc"],
            values["remote_address"],
            values["content_length"],
            values["metadata_sha256"],
            values["metadata_size"],
            values["minidump_sha256"],
            values["minidump_size"],
            values["attachments_sha256"],
            values["attachments_size"],
            values["storage_path"],
            values["metadata_json"],
            values["server_json"],
        ),
    )


def get_symbol_upload_by_identity_hash(conn, identity_hash: str):
    return conn.execute("SELECT * FROM crash_symbol_uploads WHERE identity_hash=?", (identity_hash,)).fetchone()


def get_symbol_upload_by_id(conn, symbol_upload_id: str):
    return conn.execute("SELECT * FROM crash_symbol_uploads WHERE symbol_upload_id=?", (symbol_upload_id,)).fetchone()


def insert_symbol_upload(conn, values: dict):
    conn.execute(
        """
        INSERT INTO crash_symbol_uploads(
          symbol_upload_id,identity_hash,status,product,app_version,git_commit,build_type,platform,toolchain,
          created_at_utc,received_at_utc,metadata_sha256,metadata_size,symbols_sha256,symbols_size,storage_path,metadata_json
        ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
        """,
        (
            values["symbol_upload_id"],
            values["identity_hash"],
            values["status"],
            values["product"],
            values["app_version"],
            values["git_commit"],
            values["build_type"],
            values["platform"],
            values["toolchain"],
            values["created_at_utc"],
            values["received_at_utc"],
            values["metadata_sha256"],
            values["metadata_size"],
            values["symbols_sha256"],
            values["symbols_size"],
            values["storage_path"],
            values["metadata_json"],
        ),
    )