bas.sh
Archive of https://github.com/w-social-eu/w-social-atproto (see also https://github.com/brw/w-social-atproto-archive)
Merge pull request #152 from w-social-eu/feature/label-sync-ozone-appview-main
Add label poll from ozone to appview
author
Jan Lindblad
committer
GitHub
date
(Jun 5, 2026, 11:17 AM +0200)
commit
028841c5
028841c5518d0778497b9943549f84cff5632643
parent
d0441a70
d0441a70908a26fc4019bb9d2cc440609e9fc34e
+166
-55
+9
-13
packages/bsky/src/lexicon/lexicons.ts
+9
-13
packages/bsky/src/lexicon/lexicons.ts
···
16031
16031
id: 'eu.wsocial.actor.getVerificationDetails',
16032
16032
defs: {
16033
16033
main: {
16034
-
type: 'procedure',
16034
+
type: 'query',
16035
16035
description:
16036
16036
"Fetch the merged verification record for an account. There is exactly one record per verified subject (rkey derived from DID). Returns 'NotVerified' if no record exists.",
16037
-
input: {
16038
-
encoding: 'application/json',
16039
-
schema: {
16040
-
type: 'object',
16041
-
required: ['did'],
16042
-
properties: {
16043
-
did: {
16044
-
type: 'string',
16045
-
format: 'did',
16046
-
description:
16047
-
'DID of the account to fetch verification details for.',
16048
-
},
16037
+
parameters: {
16038
+
type: 'params',
16039
+
required: ['did'],
16040
+
properties: {
16041
+
did: {
16042
+
type: 'string',
16043
+
format: 'did',
16044
+
description: 'DID of the subject to look up.',
16049
16045
},
16050
16046
},
16051
16047
},
+15
-15
pds-wadmin-modules/wadmin/commands/verification.py
+15
-15
pds-wadmin-modules/wadmin/commands/verification.py
···
31
31
# Helpers
32
32
# ---------------------------------------------------------------------------
33
33
34
+
def _is_record_not_found(resp: requests.Response) -> bool:
35
+
"""atproto PDS returns 400 with error=RecordNotFound (not 404) for missing records."""
36
+
try:
37
+
body = resp.json()
38
+
return body.get("error") in ("RecordNotFound", "NotFound")
39
+
except Exception:
40
+
return False
41
+
42
+
34
43
def _require_ozone(ctx) -> tuple[Config, OzoneClient, str]:
35
44
"""Return (config, ozone_client, labeler_pds_host) or abort if not configured."""
36
45
config: Config = ctx.obj["config"]
···
75
84
@click.option("--birth-year", "birth_year", default=None, type=int, help="True birth year (personal, WID-consented). Stored offset by a random date +/- 12 months from July 1 of that year.")
76
85
@click.option("--responsible-did", "responsible_did", default=None, help="DID of the responsible human account (for bot/org/service).")
77
86
@click.option("--approved-name", "approved_names", multiple=True, help="Approved display names (repeatable).")
78
-
@click.option(
79
-
"--method",
80
-
type=click.Choice(["admin", "wid"], case_sensitive=False),
81
-
default="admin",
82
-
show_default=True,
83
-
help="Verification method.",
84
-
)
85
87
@click.pass_context
86
88
def verify(
87
89
ctx,
···
94
96
birth_year: Optional[int],
95
97
responsible_did: Optional[str],
96
98
approved_names: tuple,
97
-
method: str,
98
99
):
99
100
"""Issue a verification label to an account (writes Ozone label + backing record)."""
100
101
config, ozone, labeler_pds_host = _require_ozone(ctx)
···
146
147
existing: dict = {}
147
148
if existing_r.status_code == 200:
148
149
existing = existing_r.json().get("value", {})
149
-
elif existing_r.status_code != 404:
150
+
elif existing_r.status_code in (404, 400) and _is_record_not_found(existing_r):
151
+
pass # first write — no existing record
152
+
else:
150
153
try:
151
154
err = existing_r.json().get("message", existing_r.text)
152
155
except Exception:
···
176
179
"createdAt": existing.get("createdAt", now),
177
180
}
178
181
179
-
if method.lower() == "admin":
180
-
record["adminVerifiedAt"] = now
181
-
record["approvedByDid"] = config.ozone_labeler_did
182
-
else:
183
-
record["widVerifiedAt"] = now
182
+
record["adminVerifiedAt"] = now
183
+
record["approvedByDid"] = config.ozone_labeler_did
184
184
185
185
if approved_names:
186
186
record["approvedDisplayNames"] = list(approved_names)
···
231
231
"$type": "tools.ozone.moderation.defs#modEventLabel",
232
232
"createLabelVals": [label_val],
233
233
"negateLabelVals": [],
234
-
"comment": f"Verified via ozone-wadmin (method: {method})",
234
+
"comment": "Verified via ozone-wadmin (method: admin)",
235
235
},
236
236
"subject": {
237
237
"$type": "com.atproto.admin.defs#repoRef",
+114
services/dataplane/index.js
+114
services/dataplane/index.js
···
8
8
9
9
// Tracer code above must come before anything else
10
10
const assert = require('node:assert')
11
+
const https = require('node:https')
12
+
const http = require('node:http')
11
13
const { Database, DataPlaneServer, RepoSubscription } = require('@atproto/bsky')
12
14
15
+
// ---------------------------------------------------------------------------
16
+
// Label poller — polls com.atproto.label.queryLabels and upserts into the
17
+
// dataplane `label` table. Cursor is persisted in the `subscription` table
18
+
// so restarts resume from where they left off.
19
+
//
20
+
// When com.atproto.label.subscribeLabels is available on the deployed Ozone
21
+
// version, swap this out for a WebSocket consumer.
22
+
// ---------------------------------------------------------------------------
23
+
24
+
const LABEL_POLL_INTERVAL_MS = 30_000
25
+
const LABEL_POLL_PAGE_SIZE = 100
26
+
27
+
async function fetchJson(url) {
28
+
return new Promise((resolve, reject) => {
29
+
const lib = url.startsWith('https') ? https : http
30
+
lib
31
+
.get(url, (res) => {
32
+
let data = ''
33
+
res.on('data', (chunk) => (data += chunk))
34
+
res.on('end', () => {
35
+
try {
36
+
resolve(JSON.parse(data))
37
+
} catch (e) {
38
+
reject(new Error(`Failed to parse JSON from ${url}: ${e.message}`))
39
+
}
40
+
})
41
+
})
42
+
.on('error', reject)
43
+
})
44
+
}
45
+
46
+
async function readLabelCursor(db, labelerUrl) {
47
+
const row = await db.db
48
+
.selectFrom('subscription')
49
+
.where('service', '=', labelerUrl)
50
+
.where('method', '=', 'queryLabels')
51
+
.select('state')
52
+
.executeTakeFirst()
53
+
return row ? row.state : '0'
54
+
}
55
+
56
+
async function writeLabelCursor(db, labelerUrl, cursor) {
57
+
await db.db
58
+
.insertInto('subscription')
59
+
.values({ service: labelerUrl, method: 'queryLabels', state: String(cursor) })
60
+
.onConflict((oc) =>
61
+
oc.columns(['service', 'method']).doUpdateSet({ state: String(cursor) }),
62
+
)
63
+
.execute()
64
+
}
65
+
66
+
async function pollLabels(db, labelerUrl, stopped) {
67
+
let cursor = await readLabelCursor(db, labelerUrl)
68
+
console.log(`label-poller: starting from cursor ${cursor} at ${labelerUrl}`)
69
+
70
+
while (!stopped.value) {
71
+
try {
72
+
const url =
73
+
`${labelerUrl}/xrpc/com.atproto.label.queryLabels` +
74
+
`?uriPatterns=*&limit=${LABEL_POLL_PAGE_SIZE}&cursor=${cursor}`
75
+
const body = await fetchJson(url)
76
+
const labels = body.labels ?? []
77
+
78
+
if (labels.length > 0) {
79
+
for (const label of labels) {
80
+
await db.db
81
+
.insertInto('label')
82
+
.values({
83
+
src: label.src,
84
+
uri: label.uri,
85
+
cid: label.cid ?? '',
86
+
val: label.val,
87
+
neg: label.neg ?? false,
88
+
cts: label.cts,
89
+
exp: label.exp ?? null,
90
+
})
91
+
.onConflict((oc) => oc.doNothing())
92
+
.execute()
93
+
}
94
+
cursor = body.cursor ?? cursor
95
+
await writeLabelCursor(db, labelerUrl, cursor)
96
+
console.log(
97
+
`label-poller: ingested ${labels.length} label(s), cursor now ${cursor}`,
98
+
)
99
+
}
100
+
} catch (err) {
101
+
console.error('label-poller: poll failed', err.message)
102
+
}
103
+
104
+
// Wait before next poll, but bail early if shutdown is requested
105
+
await new Promise((resolve) => setTimeout(resolve, LABEL_POLL_INTERVAL_MS))
106
+
}
107
+
}
108
+
13
109
const main = async () => {
14
110
const port = parseInt(process.env.DATAPLANE_PORT || '3001', 10)
15
111
const dbUrl = process.env.DATAPLANE_DB_POSTGRES_URL
···
18
114
process.env.DATAPLANE_DID_PLC_URL || 'https://plc.directory'
19
115
const poolSize = parseInt(process.env.DATAPLANE_DB_POOL_SIZE || '10', 10)
20
116
const runMigrations = process.env.DATAPLANE_DB_MIGRATE !== '0'
117
+
// Comma-separated list of labeler base URLs to poll for labels.
118
+
// Example: DATAPLANE_LABELER_URLS=https://ozone.wsocial.cloud
119
+
const labelerUrls = (process.env.DATAPLANE_LABELER_URLS || '')
120
+
.split(',')
121
+
.map((u) => u.trim())
122
+
.filter(Boolean)
21
123
22
124
// When true, start from the current tail of the firehose instead of cursor 0.
23
125
// Required for high-volume relays (bsky.network) to avoid OOM on historical backfill.
···
76
178
)
77
179
}
78
180
181
+
// Start label pollers (one per configured labeler URL)
182
+
const labelPollStopped = { value: false }
183
+
for (const labelerUrl of labelerUrls) {
184
+
pollLabels(db, labelerUrl, labelPollStopped).catch((err) => {
185
+
console.error(`label-poller: fatal error for ${labelerUrl}`, err)
186
+
})
187
+
}
188
+
if (labelerUrls.length > 0) {
189
+
console.log(`dataplane: polling labels from ${labelerUrls.join(', ')}`)
190
+
}
191
+
79
192
const shutdown = async () => {
80
193
console.log('dataplane: shutting down')
194
+
labelPollStopped.value = true
81
195
if (sub) await sub.destroy()
82
196
await dataplane.destroy()
83
197
await db.close()