Skip to content

queries

queries

Synchronizer-specific database queries.

Classes

Functions

fetch_cursors_to_sync async 

fetch_cursors_to_sync(
    brotr: Brotr, end: int, networks: Sequence[NetworkType]
) -> list[SyncCursor]

Fetch sync cursors for relays on the given networks, ordered by timestamp ascending.

Performs a LEFT JOIN between the relay table and the synchronizer's cursor state in service_state. Only relays whose network is in networks are included. Relays without a cursor get a default cursor (timestamp=0). Relays already synced past end are excluded.

Parameters:

  • brotr (Brotr) –

    Brotr database interface.

  • end (int) –

    Upper bound timestamp — relays synced past this are excluded.

  • networks (Sequence[NetworkType]) –

    Network types to include (e.g. enabled networks from config).

Returns:

Source code in src/bigbrotr/services/synchronizer/queries.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
async def fetch_cursors_to_sync(
    brotr: Brotr, end: int, networks: Sequence[NetworkType]
) -> list[SyncCursor]:
    """Fetch sync cursors for relays on the given networks, ordered by timestamp ascending.

    Performs a LEFT JOIN between the relay table and the synchronizer's
    cursor state in ``service_state``. Only relays whose network is in
    ``networks`` are included. Relays without a cursor get a default
    cursor (``timestamp=0``). Relays already synced past ``end`` are excluded.

    Args:
        brotr: [Brotr][bigbrotr.core.brotr.Brotr] database interface.
        end: Upper bound timestamp — relays synced past this are excluded.
        networks: Network types to include (e.g. enabled networks from config).

    Returns:
        List of [SyncCursor][bigbrotr.services.common.types.SyncCursor]
        ordered by timestamp ascending (oldest first).
    """
    rows = await brotr.fetch(
        """
        WITH cursors AS (
            SELECT state_key,
                   state_value,
                   (state_value->>'timestamp')::bigint AS ts,
                   state_value->>'id' AS cursor_id
            FROM service_state
            WHERE service_name = $1
              AND state_type = $2
        )
        SELECT r.url, c.state_value
        FROM relay r
        LEFT JOIN cursors c ON c.state_key = r.url
        WHERE r.network = ANY($4)
          AND (c.ts IS NULL OR c.ts <= $3)
        ORDER BY COALESCE(c.ts, 0) ASC,
                 COALESCE(c.cursor_id, repeat('0', 64)) ASC
        """,
        ServiceName.SYNCHRONIZER,
        ServiceStateType.CURSOR,
        end,
        [n.value for n in networks],
    )
    results: list[SyncCursor] = []
    for row in rows:
        sv = row["state_value"]
        if sv:
            results.append(
                SyncCursor(key=row["url"], timestamp=int(sv["timestamp"]), id=str(sv["id"]))
            )
        else:
            results.append(SyncCursor(key=row["url"]))
    return results

delete_stale_cursors async 

delete_stale_cursors(brotr: Brotr) -> int

Remove cursor state for relays that no longer exist.

Deletes CURSOR records whose state_key does not match any relay in the relay table.

Parameters:

Returns:

  • int –

    Number of deleted rows.

Source code in src/bigbrotr/services/synchronizer/queries.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
async def delete_stale_cursors(brotr: Brotr) -> int:
    """Remove cursor state for relays that no longer exist.

    Deletes CURSOR records whose ``state_key`` does not match any relay in
    the relay table.

    Args:
        brotr: [Brotr][bigbrotr.core.brotr.Brotr] database interface.

    Returns:
        Number of deleted rows.
    """
    count: int = await brotr.fetchval(
        """
        WITH deleted AS (
            DELETE FROM service_state
            WHERE service_name = $1
              AND state_type = $2
              AND NOT EXISTS (SELECT 1 FROM relay r WHERE r.url = state_key)
            RETURNING 1
        )
        SELECT count(*)::int FROM deleted
        """,
        ServiceName.SYNCHRONIZER,
        ServiceStateType.CURSOR,
    )
    return count

insert_event_relays async 

insert_event_relays(
    brotr: Brotr, records: list[EventRelay]
) -> int

Batch-insert event-relay junction records.

Splits records into batches respecting BatchConfig.max_size and delegates each chunk to Brotr.insert_event_relay() (cascade mode).

Parameters:

Returns:

  • int –

    Number of new event-relay records inserted.

Source code in src/bigbrotr/services/synchronizer/queries.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
async def insert_event_relays(brotr: Brotr, records: list[EventRelay]) -> int:
    """Batch-insert event-relay junction records.

    Splits *records* into batches respecting
    [BatchConfig.max_size][bigbrotr.core.brotr.BatchConfig] and delegates
    each chunk to
    [Brotr.insert_event_relay()][bigbrotr.core.brotr.Brotr.insert_event_relay]
    (cascade mode).

    Args:
        brotr: [Brotr][bigbrotr.core.brotr.Brotr] database interface.
        records: [EventRelay][bigbrotr.models.event_relay.EventRelay]
            instances to insert.

    Returns:
        Number of new event-relay records inserted.
    """
    return await batched_insert(brotr, records, brotr.insert_event_relay)

upsert_sync_cursors async 

upsert_sync_cursors(
    brotr: Brotr, cursors: Iterable[SyncCursor]
) -> None

Persist multiple sync cursor positions in batched upserts.

Splits cursors into chunks of brotr.config.batch.max_size to respect database parameter limits.

Parameters:

Source code in src/bigbrotr/services/synchronizer/queries.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
async def upsert_sync_cursors(brotr: Brotr, cursors: Iterable[SyncCursor]) -> None:
    """Persist multiple sync cursor positions in batched upserts.

    Splits cursors into chunks of ``brotr.config.batch.max_size`` to
    respect database parameter limits.

    Args:
        brotr: [Brotr][bigbrotr.core.brotr.Brotr] database interface.
        cursors: Iterable of [SyncCursor][bigbrotr.services.common.types.SyncCursor]
            instances to persist.
    """
    states = [
        ServiceState(
            service_name=ServiceName.SYNCHRONIZER,
            state_type=ServiceStateType.CURSOR,
            state_key=cursor.key,
            state_value={
                "timestamp": cursor.timestamp,
                "id": cursor.id,
            },
        )
        for cursor in cursors
    ]
    await upsert_service_states(brotr, states)