OXYGENHUB
# b0612e7dce30 oxygenhub / crates / hub / src / db / accounts.rs
28.9 KiB · Rust · last changed Device tokens expire and renew on use; the approval page shows the requester (closes #70) 1 month ago source blame history raw
1 
//! Accounts, organisations, teams, collaborators, identities, tokens,
2 
//! invitations, settings and the audit log (hub spec §5).
3
4 
use super::{now_ms, Pool};
5 
use sqlx::FromRow;
6
7 
#[derive(Clone, Debug, FromRow)]
8 
pub struct Account {
9 
    pub id: i64,
10 
    pub username: String,
11 
    pub display_name: String,
12 
    pub email: String,
13 
    pub email_verified: bool,
14 
    pub password_hash: String,
15 
    pub session_salt: String,
16 
    pub totp_secret: Option<String>,
17 
    pub recovery_codes: String,
18 
    pub is_admin: bool,
19 
    pub banned: bool,
20 
    pub created_ms: i64,
21 
    pub last_seen_ms: i64,
22 
    pub email_notifications: bool,
23 
    pub unsubscribe_key: String,
24 
    /// Shown on the profile page (migration 0008).
25 
    #[sqlx(default)]
26 
    pub bio: String,
27 
    #[sqlx(default)]
28 
    pub website: String,
29 
    #[sqlx(default)]
30 
    pub location: String,
31 
    /// `png:<hash>` or `jpeg:<hash>` when a picture was uploaded.
32 
    pub avatar: String,
33 
}
34
35 
impl Account {
36 
    pub fn name(&self) -> &str {
37 
        if self.display_name.is_empty() {
38 
            &self.username
39 
        } else {
40 
            &self.display_name
41 
        }
42 
    }
43
44 
    pub fn totp_enabled(&self) -> bool {
45 
        self.totp_secret.is_some()
46 
    }
47 
}
48
49 
#[derive(Clone, Debug, FromRow)]
50 
pub struct Organisation {
51 
    pub id: i64,
52 
    pub name: String,
53 
    pub display_name: String,
54 
    pub description: String,
55 
    pub default_role: String,
56 
    pub created_ms: i64,
57 
    /// `png:<hash>` or `jpeg:<hash>` when a picture was uploaded.
58 
    pub avatar: String,
59 
}
60
61 
#[derive(Clone, Debug, FromRow)]
62 
pub struct Team {
63 
    pub id: i64,
64 
    pub org_id: i64,
65 
    pub name: String,
66 
}
67
68 
#[derive(Clone, Debug, FromRow)]
69 
pub struct Token {
70 
    pub id: i64,
71 
    pub account_id: i64,
72 
    pub name: String,
73 
    pub hash: String,
74 
    pub scopes: String,
75 
    pub repos: String,
76 
    pub created_ms: i64,
77 
    pub expires_ms: Option<i64>,
78 
    pub last_used_ms: Option<i64>,
79 
    /// A renewing token's lifetime: each use extends it to now plus this.
80 
    pub renew_ms: Option<i64>,
81 
}
82
83 
impl Token {
84 
    pub fn scope_list(&self) -> Vec<&str> {
85 
        self.scopes.split(',').filter(|s| !s.is_empty()).collect()
86 
    }
87
88 
    pub fn has_scope(&self, scope: &str) -> bool {
89 
        self.scope_list().contains(&scope)
90 
    }
91
92 
    /// Whether the token may touch `owner/name`: every repository when the
93 
    /// list is empty.
94 
    pub fn covers(&self, slug: &str) -> bool {
95 
        self.repos.is_empty() || self.repos.split(',').any(|r| r.trim().eq_ignore_ascii_case(slug))
96 
    }
97
98 
    pub fn expired(&self, now: i64) -> bool {
99 
        self.expires_ms.is_some_and(|e| e < now)
100 
    }
101 
}
102
103 
#[derive(Clone, Debug, FromRow)]
104 
pub struct LinkedIdentity {
105 
    pub key: String,
106 
    pub account_id: i64,
107 
    pub name: String,
108 
    pub email: String,
109 
    pub object: Vec<u8>,
110 
    pub linked_ms: i64,
111 
}
112
113 
#[derive(Clone, Debug, FromRow)]
114 
pub struct AuditRow {
115 
    pub id: i64,
116 
    pub at_ms: i64,
117 
    pub actor: String,
118 
    pub action: String,
119 
    pub target: String,
120 
    pub detail: String,
121 
}
122
123 
// ---------------------------------------------------------------- accounts
124
125 
pub async fn account_by_name(pool: &Pool, username: &str) -> sqlx::Result<Option<Account>> {
126 
    sqlx::query_as("SELECT * FROM accounts WHERE username = ?").bind(username).fetch_optional(pool).await
127 
}
128
129 
pub async fn account_by_id(pool: &Pool, id: i64) -> sqlx::Result<Option<Account>> {
130 
    sqlx::query_as("SELECT * FROM accounts WHERE id = ?").bind(id).fetch_optional(pool).await
131 
}
132
133 
pub async fn accounts(pool: &Pool) -> sqlx::Result<Vec<Account>> {
134 
    sqlx::query_as("SELECT * FROM accounts ORDER BY username").fetch_all(pool).await
135 
}
136
137 
pub async fn account_count(pool: &Pool) -> sqlx::Result<i64> {
138 
    sqlx::query_scalar("SELECT COUNT(*) FROM accounts").fetch_one(pool).await
139 
}
140
141 
pub async fn create_account(
142 
    pool: &Pool,
143 
    username: &str,
144 
    email: &str,
145 
    password_hash: &str,
146 
    session_salt: &str,
147 
    is_admin: bool,
148 
) -> sqlx::Result<Account> {
149 
    sqlx::query_as(
150 
        "INSERT INTO accounts (username, email, password_hash, session_salt, is_admin, created_ms) VALUES (?, ?, ?, ?, ?, ?) RETURNING *",
151 
    )
152 
    .bind(username)
153 
    .bind(email)
154 
    .bind(password_hash)
155 
    .bind(session_salt)
156 
    .bind(is_admin)
157 
    .bind(now_ms())
158 
    .fetch_one(pool)
159 
    .await
160 
}
161
162 
pub async fn set_profile(pool: &Pool, id: i64, display_name: &str, email: &str) -> sqlx::Result<()> {
163 
    // A changed address is unverified until its own link is used.
164 
    sqlx::query(
165 
        "UPDATE accounts SET display_name = ?, email_verified = CASE WHEN email = ? THEN email_verified ELSE 0 END, email = ? WHERE id = ?",
166 
    )
167 
    .bind(display_name)
168 
    .bind(email)
169 
    .bind(email)
170 
    .bind(id)
171 
    .execute(pool)
172 
    .await?;
173 
    Ok(())
174 
}
175
176 
/// The public profile fields.
177 
pub async fn set_bio(pool: &Pool, id: i64, bio: &str, website: &str, location: &str) -> sqlx::Result<()> {
178 
    sqlx::query("UPDATE accounts SET bio = ?, website = ?, location = ? WHERE id = ?")
179 
        .bind(bio)
180 
        .bind(website)
181 
        .bind(location)
182 
        .bind(id)
183 
        .execute(pool)
184 
        .await?;
185 
    Ok(())
186 
}
187
188 
pub async fn set_password(pool: &Pool, id: i64, password_hash: &str, session_salt: &str) -> sqlx::Result<()> {
189 
    sqlx::query("UPDATE accounts SET password_hash = ?, session_salt = ? WHERE id = ?")
190 
        .bind(password_hash)
191 
        .bind(session_salt)
192 
        .bind(id)
193 
        .execute(pool)
194 
        .await?;
195 
    Ok(())
196 
}
197
198 
pub async fn set_session_salt(pool: &Pool, id: i64, session_salt: &str) -> sqlx::Result<()> {
199 
    sqlx::query("UPDATE accounts SET session_salt = ? WHERE id = ?").bind(session_salt).bind(id).execute(pool).await?;
200 
    Ok(())
201 
}
202
203 
pub async fn set_totp(pool: &Pool, id: i64, secret: Option<&str>, recovery_codes: &str) -> sqlx::Result<()> {
204 
    sqlx::query("UPDATE accounts SET totp_secret = ?, recovery_codes = ? WHERE id = ?")
205 
        .bind(secret)
206 
        .bind(recovery_codes)
207 
        .bind(id)
208 
        .execute(pool)
209 
        .await?;
210 
    Ok(())
211 
}
212
213 
pub async fn set_admin(pool: &Pool, id: i64, is_admin: bool) -> sqlx::Result<()> {
214 
    sqlx::query("UPDATE accounts SET is_admin = ? WHERE id = ?").bind(is_admin).bind(id).execute(pool).await?;
215 
    Ok(())
216 
}
217
218 
pub async fn set_banned(pool: &Pool, id: i64, banned: bool) -> sqlx::Result<()> {
219 
    sqlx::query("UPDATE accounts SET banned = ? WHERE id = ?").bind(banned).bind(id).execute(pool).await?;
220 
    Ok(())
221 
}
222
223 
pub async fn touch_seen(pool: &Pool, id: i64) -> sqlx::Result<()> {
224 
    sqlx::query("UPDATE accounts SET last_seen_ms = ? WHERE id = ?").bind(now_ms()).bind(id).execute(pool).await?;
225 
    Ok(())
226 
}
227
228 
// ----------------------------------------------------------- organisations
229
230 
pub async fn org_by_name(pool: &Pool, name: &str) -> sqlx::Result<Option<Organisation>> {
231 
    sqlx::query_as("SELECT * FROM organisations WHERE name = ?").bind(name).fetch_optional(pool).await
232 
}
233
234 
pub async fn orgs(pool: &Pool) -> sqlx::Result<Vec<Organisation>> {
235 
    sqlx::query_as("SELECT * FROM organisations ORDER BY name").fetch_all(pool).await
236 
}
237
238 
/// An organisation with the account's role in it.
239 
#[derive(FromRow)]
240 
struct Membership {
241 
    #[sqlx(flatten)]
242 
    org: Organisation,
243 
    role: String,
244 
}
245
246 
pub async fn orgs_of(pool: &Pool, account_id: i64) -> sqlx::Result<Vec<(Organisation, String)>> {
247 
    let rows: Vec<Membership> = sqlx::query_as(
248 
        "SELECT o.*, m.role FROM organisations o JOIN org_members m ON m.org_id = o.id WHERE m.account_id = ? ORDER BY o.name",
249 
    )
250 
    .bind(account_id)
251 
    .fetch_all(pool)
252 
    .await?;
253 
    Ok(rows.into_iter().map(|m| (m.org, m.role)).collect())
254 
}
255
256 
pub async fn create_org(pool: &Pool, name: &str, display_name: &str, owner: i64) -> sqlx::Result<Organisation> {
257 
    let mut tx = super::begin_write(pool).await?;
258 
    let org: Organisation = sqlx::query_as("INSERT INTO organisations (name, display_name, created_ms) VALUES (?, ?, ?) RETURNING *")
259 
        .bind(name)
260 
        .bind(display_name)
261 
        .bind(now_ms())
262 
        .fetch_one(&mut *tx)
263 
        .await?;
264 
    sqlx::query("INSERT INTO org_members (org_id, account_id, role) VALUES (?, ?, 'owner')")
265 
        .bind(org.id)
266 
        .bind(owner)
267 
        .execute(&mut *tx)
268 
        .await?;
269 
    tx.commit().await?;
270 
    Ok(org)
271 
}
272
273 
pub async fn set_org(pool: &Pool, id: i64, display_name: &str, description: &str, default_role: &str) -> sqlx::Result<()> {
274 
    sqlx::query("UPDATE organisations SET display_name = ?, description = ?, default_role = ? WHERE id = ?")
275 
        .bind(display_name)
276 
        .bind(description)
277 
        .bind(default_role)
278 
        .bind(id)
279 
        .execute(pool)
280 
        .await?;
281 
    Ok(())
282 
}
283
284 
/// `(account, role)` for every member.
285 
pub async fn org_members(pool: &Pool, org_id: i64) -> sqlx::Result<Vec<(Account, String)>> {
286 
    let rows: Vec<(i64, String)> =
287 
        sqlx::query_as("SELECT account_id, role FROM org_members WHERE org_id = ?").bind(org_id).fetch_all(pool).await?;
288 
    let mut out = Vec::new();
289 
    for (id, role) in rows {
290 
        if let Some(a) = account_by_id(pool, id).await? {
291 
            out.push((a, role));
292 
        }
293 
    }
294 
    out.sort_by(|a, b| a.0.username.cmp(&b.0.username));
295 
    Ok(out)
296 
}
297
298 
pub async fn org_role(pool: &Pool, org_id: i64, account_id: i64) -> sqlx::Result<Option<String>> {
299 
    sqlx::query_scalar("SELECT role FROM org_members WHERE org_id = ? AND account_id = ?")
300 
        .bind(org_id)
301 
        .bind(account_id)
302 
        .fetch_optional(pool)
303 
        .await
304 
}
305
306 
pub async fn set_org_member(pool: &Pool, org_id: i64, account_id: i64, role: &str) -> sqlx::Result<()> {
307 
    sqlx::query("INSERT INTO org_members (org_id, account_id, role) VALUES (?, ?, ?) ON CONFLICT (org_id, account_id) DO UPDATE SET role = excluded.role")
308 
        .bind(org_id)
309 
        .bind(account_id)
310 
        .bind(role)
311 
        .execute(pool)
312 
        .await?;
313 
    Ok(())
314 
}
315
316 
pub async fn remove_org_member(pool: &Pool, org_id: i64, account_id: i64) -> sqlx::Result<()> {
317 
    sqlx::query("DELETE FROM org_members WHERE org_id = ? AND account_id = ?").bind(org_id).bind(account_id).execute(pool).await?;
318 
    Ok(())
319 
}
320
321 
pub async fn teams(pool: &Pool, org_id: i64) -> sqlx::Result<Vec<Team>> {
322 
    sqlx::query_as("SELECT * FROM teams WHERE org_id = ? ORDER BY name").bind(org_id).fetch_all(pool).await
323 
}
324
325 
pub async fn team_by_name(pool: &Pool, org_id: i64, name: &str) -> sqlx::Result<Option<Team>> {
326 
    sqlx::query_as("SELECT * FROM teams WHERE org_id = ? AND name = ?").bind(org_id).bind(name).fetch_optional(pool).await
327 
}
328
329 
pub async fn create_team(pool: &Pool, org_id: i64, name: &str) -> sqlx::Result<Team> {
330 
    sqlx::query_as("INSERT INTO teams (org_id, name) VALUES (?, ?) RETURNING *").bind(org_id).bind(name).fetch_one(pool).await
331 
}
332
333 
pub async fn delete_team(pool: &Pool, id: i64) -> sqlx::Result<()> {
334 
    sqlx::query("DELETE FROM teams WHERE id = ?").bind(id).execute(pool).await?;
335 
    Ok(())
336 
}
337
338 
pub async fn team_members(pool: &Pool, team_id: i64) -> sqlx::Result<Vec<Account>> {
339 
    sqlx::query_as("SELECT a.* FROM accounts a JOIN team_members t ON t.account_id = a.id WHERE t.team_id = ? ORDER BY a.username")
340 
        .bind(team_id)
341 
        .fetch_all(pool)
342 
        .await
343 
}
344
345 
pub async fn add_team_member(pool: &Pool, team_id: i64, account_id: i64) -> sqlx::Result<()> {
346 
    sqlx::query("INSERT OR IGNORE INTO team_members (team_id, account_id) VALUES (?, ?)")
347 
        .bind(team_id)
348 
        .bind(account_id)
349 
        .execute(pool)
350 
        .await?;
351 
    Ok(())
352 
}
353
354 
pub async fn remove_team_member(pool: &Pool, team_id: i64, account_id: i64) -> sqlx::Result<()> {
355 
    sqlx::query("DELETE FROM team_members WHERE team_id = ? AND account_id = ?").bind(team_id).bind(account_id).execute(pool).await?;
356 
    Ok(())
357 
}
358
359 
// ----------------------------------------------------------- collaborators
360
361 
#[derive(Clone, Debug)]
362 
pub struct Collaborator {
363 
    pub account: Option<Account>,
364 
    pub team: Option<Team>,
365 
    pub role: String,
366 
}
367
368 
pub async fn collaborators(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<Collaborator>> {
369 
    let rows: Vec<(Option<i64>, Option<i64>, String)> =
370 
        sqlx::query_as("SELECT account_id, team_id, role FROM collaborators WHERE repo_id = ?").bind(repo_id).fetch_all(pool).await?;
371 
    let mut out = Vec::new();
372 
    for (account_id, team_id, role) in rows {
373 
        let account = match account_id {
374 
            Some(id) => account_by_id(pool, id).await?,
375 
            None => None,
376 
        };
377 
        let team = match team_id {
378 
            Some(id) => sqlx::query_as("SELECT * FROM teams WHERE id = ?").bind(id).fetch_optional(pool).await?,
379 
            None => None,
380 
        };
381 
        out.push(Collaborator { account, team, role });
382 
    }
383 
    Ok(out)
384 
}
385
386 
pub async fn set_collaborator(pool: &Pool, repo_id: i64, account_id: Option<i64>, team_id: Option<i64>, role: &str) -> sqlx::Result<()> {
387 
    if let Some(a) = account_id {
388 
        sqlx::query("INSERT INTO collaborators (repo_id, account_id, role) VALUES (?, ?, ?) ON CONFLICT (repo_id, account_id) WHERE account_id IS NOT NULL DO UPDATE SET role = excluded.role")
389 
            .bind(repo_id)
390 
            .bind(a)
391 
            .bind(role)
392 
            .execute(pool)
393 
            .await?;
394 
    } else if let Some(t) = team_id {
395 
        sqlx::query("INSERT INTO collaborators (repo_id, team_id, role) VALUES (?, ?, ?) ON CONFLICT (repo_id, team_id) WHERE team_id IS NOT NULL DO UPDATE SET role = excluded.role")
396 
            .bind(repo_id)
397 
            .bind(t)
398 
            .bind(role)
399 
            .execute(pool)
400 
            .await?;
401 
    }
402 
    Ok(())
403 
}
404
405 
pub async fn remove_collaborator(pool: &Pool, repo_id: i64, account_id: Option<i64>, team_id: Option<i64>) -> sqlx::Result<()> {
406 
    sqlx::query("DELETE FROM collaborators WHERE repo_id = ? AND ((account_id IS NOT NULL AND account_id = ?) OR (team_id IS NOT NULL AND team_id = ?))")
407 
        .bind(repo_id)
408 
        .bind(account_id)
409 
        .bind(team_id)
410 
        .execute(pool)
411 
        .await?;
412 
    Ok(())
413 
}
414
415 
/// The direct or team role an account holds on a repository, the highest.
416 
pub async fn collaborator_role(pool: &Pool, repo_id: i64, account_id: i64) -> sqlx::Result<Option<String>> {
417 
    let rows: Vec<String> = sqlx::query_scalar(
418 
        "SELECT role FROM collaborators WHERE repo_id = ? AND (account_id = ? OR team_id IN (SELECT team_id FROM team_members WHERE account_id = ?))",
419 
    )
420 
    .bind(repo_id)
421 
    .bind(account_id)
422 
    .bind(account_id)
423 
    .fetch_all(pool)
424 
    .await?;
425 
    Ok(rows.into_iter().max_by_key(|r| crate::auth::Role::parse(r).map(|r| r as u8).unwrap_or(0)))
426 
}
427
428 
// ------------------------------------------------------------------ tokens
429
430 
pub async fn tokens_of(pool: &Pool, account_id: i64) -> sqlx::Result<Vec<Token>> {
431 
    sqlx::query_as("SELECT * FROM tokens WHERE account_id = ? ORDER BY created_ms DESC").bind(account_id).fetch_all(pool).await
432 
}
433
434 
pub async fn token_by_hash(pool: &Pool, hash: &str) -> sqlx::Result<Option<Token>> {
435 
    sqlx::query_as("SELECT * FROM tokens WHERE hash = ?").bind(hash).fetch_optional(pool).await
436 
}
437
438 
pub async fn create_token(
439 
    pool: &Pool,
440 
    account_id: i64,
441 
    name: &str,
442 
    hash: &str,
443 
    scopes: &str,
444 
    repos: &str,
445 
    expires_ms: Option<i64>,
446 
) -> sqlx::Result<Token> {
447 
    sqlx::query_as(
448 
        "INSERT INTO tokens (account_id, name, hash, scopes, repos, created_ms, expires_ms) VALUES (?, ?, ?, ?, ?, ?, ?) RETURNING *",
449 
    )
450 
    .bind(account_id)
451 
    .bind(name)
452 
    .bind(hash)
453 
    .bind(scopes)
454 
    .bind(repos)
455 
    .bind(now_ms())
456 
    .bind(expires_ms)
457 
    .fetch_one(pool)
458 
    .await
459 
}
460
461 
pub async fn delete_token(pool: &Pool, account_id: i64, id: i64) -> sqlx::Result<()> {
462 
    sqlx::query("DELETE FROM tokens WHERE id = ? AND account_id = ?").bind(id).bind(account_id).execute(pool).await?;
463 
    Ok(())
464 
}
465
466 
/// Record a use; a renewing token's expiry moves to now plus its lifetime.
467 
pub async fn touch_token(pool: &Pool, id: i64) -> sqlx::Result<()> {
468 
    sqlx::query(
469 
        "UPDATE tokens SET last_used_ms = ?1, \
470 
         expires_ms = CASE WHEN renew_ms IS NULL THEN expires_ms ELSE MAX(COALESCE(expires_ms, 0), ?1 + renew_ms) END \
471 
         WHERE id = ?2",
472 
    )
473 
    .bind(now_ms())
474 
    .bind(id)
475 
    .execute(pool)
476 
    .await?;
477 
    Ok(())
478 
}
479
480 
/// Make a token renewing: it lasts `renew_ms` past its last use.
481 
pub async fn set_token_renewal(pool: &Pool, id: i64, renew_ms: i64) -> sqlx::Result<()> {
482 
    sqlx::query("UPDATE tokens SET renew_ms = ? WHERE id = ?").bind(renew_ms).bind(id).execute(pool).await?;
483 
    Ok(())
484 
}
485
486 
// ------------------------------------------------------- device sign-in
487
488 
/// A device code an app asked for (oxygenhub#67): pending until the user
489 
/// approves or denies it on the web.
490 
#[derive(Clone, Debug, FromRow)]
491 
pub struct DeviceCode {
492 
    pub id: i64,
493 
    pub device_hash: String,
494 
    pub user_code: String,
495 
    pub client: String,
496 
    pub scopes: String,
497 
    pub created_ms: i64,
498 
    pub expires_ms: i64,
499 
    pub last_poll_ms: i64,
500 
    pub account_id: Option<i64>,
501 
    pub approved_scopes: String,
502 
    pub denied: bool,
503 
    /// The address that asked for the code, as the hub saw it.
504 
    pub requester: String,
505 
}
506
507 
impl DeviceCode {
508 
    pub fn scope_list(&self) -> Vec<&str> {
509 
        self.scopes.split(',').filter(|s| !s.is_empty()).collect()
510 
    }
511 
}
512
513 
pub async fn create_device_code(
514 
    pool: &Pool,
515 
    device_hash: &str,
516 
    user_code: &str,
517 
    client: &str,
518 
    scopes: &str,
519 
    expires_ms: i64,
520 
    requester: &str,
521 
) -> sqlx::Result<DeviceCode> {
522 
    // Expired codes go whenever a new one is made.
523 
    sqlx::query("DELETE FROM device_codes WHERE expires_ms < ?").bind(now_ms()).execute(pool).await?;
524 
    sqlx::query_as(
525 
        "INSERT INTO device_codes (device_hash, user_code, client, scopes, created_ms, expires_ms, requester) VALUES (?, ?, ?, ?, ?, ?, ?) RETURNING *",
526 
    )
527 
    .bind(device_hash)
528 
    .bind(user_code)
529 
    .bind(client)
530 
    .bind(scopes)
531 
    .bind(now_ms())
532 
    .bind(expires_ms)
533 
    .bind(requester)
534 
    .fetch_one(pool)
535 
    .await
536 
}
537
538 
pub async fn device_by_user_code(pool: &Pool, user_code: &str) -> sqlx::Result<Option<DeviceCode>> {
539 
    sqlx::query_as("SELECT * FROM device_codes WHERE user_code = ?").bind(user_code).fetch_optional(pool).await
540 
}
541
542 
pub async fn device_by_hash(pool: &Pool, device_hash: &str) -> sqlx::Result<Option<DeviceCode>> {
543 
    sqlx::query_as("SELECT * FROM device_codes WHERE device_hash = ?").bind(device_hash).fetch_optional(pool).await
544 
}
545
546 
/// Approve a pending code for an account with these scopes; false when it
547 
/// was no longer pending.
548 
pub async fn approve_device(pool: &Pool, id: i64, account_id: i64, scopes: &str) -> sqlx::Result<bool> {
549 
    let r = sqlx::query(
550 
        "UPDATE device_codes SET account_id = ?, approved_scopes = ? WHERE id = ? AND account_id IS NULL AND denied = 0 AND expires_ms >= ?",
551 
    )
552 
    .bind(account_id)
553 
    .bind(scopes)
554 
    .bind(id)
555 
    .bind(now_ms())
556 
    .execute(pool)
557 
    .await?;
558 
    Ok(r.rows_affected() == 1)
559 
}
560
561 
pub async fn deny_device(pool: &Pool, id: i64) -> sqlx::Result<()> {
562 
    sqlx::query("UPDATE device_codes SET denied = 1 WHERE id = ? AND account_id IS NULL").bind(id).execute(pool).await?;
563 
    Ok(())
564 
}
565
566 
pub async fn device_polled(pool: &Pool, id: i64) -> sqlx::Result<()> {
567 
    sqlx::query("UPDATE device_codes SET last_poll_ms = ? WHERE id = ?").bind(now_ms()).bind(id).execute(pool).await?;
568 
    Ok(())
569 
}
570
571 
/// Remove a code; true when this call removed it (so one poll gets the
572 
/// token, however many race).
573 
pub async fn delete_device(pool: &Pool, id: i64) -> sqlx::Result<bool> {
574 
    let r = sqlx::query("DELETE FROM device_codes WHERE id = ?").bind(id).execute(pool).await?;
575 
    Ok(r.rows_affected() == 1)
576 
}
577
578 
// -------------------------------------------------------------- identities
579
580 
pub async fn identities_of(pool: &Pool, account_id: i64) -> sqlx::Result<Vec<LinkedIdentity>> {
581 
    sqlx::query_as("SELECT * FROM identities WHERE account_id = ? ORDER BY linked_ms DESC").bind(account_id).fetch_all(pool).await
582 
}
583
584 
pub async fn identity_by_key(pool: &Pool, key: &str) -> sqlx::Result<Option<LinkedIdentity>> {
585 
    sqlx::query_as("SELECT * FROM identities WHERE key = ?").bind(key).fetch_optional(pool).await
586 
}
587
588 
/// Every linked key with its account's username, for attribution.
589 
pub async fn identity_accounts(pool: &Pool) -> sqlx::Result<Vec<(String, String)>> {
590 
    sqlx::query_as("SELECT i.key, a.username FROM identities i JOIN accounts a ON a.id = i.account_id").fetch_all(pool).await
591 
}
592
593 
pub async fn link_identity(pool: &Pool, account_id: i64, key: &str, name: &str, email: &str, object: &[u8]) -> sqlx::Result<()> {
594 
    sqlx::query("INSERT INTO identities (key, account_id, name, email, object, linked_ms) VALUES (?, ?, ?, ?, ?, ?)")
595 
        .bind(key)
596 
        .bind(account_id)
597 
        .bind(name)
598 
        .bind(email)
599 
        .bind(object)
600 
        .bind(now_ms())
601 
        .execute(pool)
602 
        .await?;
603 
    Ok(())
604 
}
605
606 
pub async fn unlink_identity(pool: &Pool, account_id: i64, key: &str) -> sqlx::Result<()> {
607 
    sqlx::query("DELETE FROM identities WHERE key = ? AND account_id = ?").bind(key).bind(account_id).execute(pool).await?;
608 
    Ok(())
609 
}
610
611 
pub async fn create_challenge(pool: &Pool, account_id: i64, bytes_hex: &str) -> sqlx::Result<i64> {
612 
    sqlx::query("DELETE FROM challenges WHERE created_ms < ?").bind(now_ms() - 10 * 60 * 1000).execute(pool).await?;
613 
    let id: i64 = sqlx::query_scalar("INSERT INTO challenges (account_id, bytes, created_ms) VALUES (?, ?, ?) RETURNING id")
614 
        .bind(account_id)
615 
        .bind(bytes_hex)
616 
        .bind(now_ms())
617 
        .fetch_one(pool)
618 
        .await?;
619 
    Ok(id)
620 
}
621
622 
/// Take (and delete) a pending challenge of this account.
623 
pub async fn take_challenge(pool: &Pool, account_id: i64, id: i64) -> sqlx::Result<Option<String>> {
624 
    let bytes: Option<String> = sqlx::query_scalar("SELECT bytes FROM challenges WHERE id = ? AND account_id = ? AND created_ms >= ?")
625 
        .bind(id)
626 
        .bind(account_id)
627 
        .bind(now_ms() - 10 * 60 * 1000)
628 
        .fetch_optional(pool)
629 
        .await?;
630 
    sqlx::query("DELETE FROM challenges WHERE id = ?").bind(id).execute(pool).await?;
631 
    Ok(bytes)
632 
}
633
634 
// ------------------------------------------------------------- invitations
635
636 
pub async fn create_invitation(pool: &Pool, code: &str, created_by: Option<i64>) -> sqlx::Result<()> {
637 
    sqlx::query("INSERT INTO invitations (code, created_by, created_ms) VALUES (?, ?, ?)")
638 
        .bind(code)
639 
        .bind(created_by)
640 
        .bind(now_ms())
641 
        .execute(pool)
642 
        .await?;
643 
    Ok(())
644 
}
645
646 
pub async fn invitation_unused(pool: &Pool, code: &str) -> sqlx::Result<bool> {
647 
    let n: i64 =
648 
        sqlx::query_scalar("SELECT COUNT(*) FROM invitations WHERE code = ? AND used_by IS NULL").bind(code).fetch_one(pool).await?;
649 
    Ok(n > 0)
650 
}
651
652 
pub async fn use_invitation(pool: &Pool, code: &str, account_id: i64) -> sqlx::Result<()> {
653 
    sqlx::query("UPDATE invitations SET used_by = ? WHERE code = ?").bind(account_id).bind(code).execute(pool).await?;
654 
    Ok(())
655 
}
656
657 
pub async fn invitations(pool: &Pool) -> sqlx::Result<Vec<(String, Option<i64>, i64)>> {
658 
    sqlx::query_as("SELECT code, used_by, created_ms FROM invitations ORDER BY created_ms DESC").fetch_all(pool).await
659 
}
660
661 
// ---------------------------------------------------------------- settings
662
663 
pub async fn setting(pool: &Pool, key: &str) -> sqlx::Result<Option<String>> {
664 
    sqlx::query_scalar("SELECT value FROM settings WHERE key = ?").bind(key).fetch_optional(pool).await
665 
}
666
667 
pub async fn set_setting(pool: &Pool, key: &str, value: &str) -> sqlx::Result<()> {
668 
    sqlx::query("INSERT INTO settings (key, value) VALUES (?, ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value")
669 
        .bind(key)
670 
        .bind(value)
671 
        .execute(pool)
672 
        .await?;
673 
    Ok(())
674 
}
675
676 
// ------------------------------------------------------------------- audit
677
678 
pub async fn audit(pool: &Pool, actor: Option<&Account>, action: &str, target: &str, detail: &str) -> sqlx::Result<()> {
679 
    sqlx::query("INSERT INTO audit (at_ms, actor_id, actor, action, target, detail) VALUES (?, ?, ?, ?, ?, ?)")
680 
        .bind(now_ms())
681 
        .bind(actor.map(|a| a.id))
682 
        .bind(actor.map(|a| a.username.as_str()).unwrap_or(""))
683 
        .bind(action)
684 
        .bind(target)
685 
        .bind(detail)
686 
        .execute(pool)
687 
        .await?;
688 
    Ok(())
689 
}
690
691 
pub async fn audit_rows(pool: &Pool, limit: i64) -> sqlx::Result<Vec<AuditRow>> {
692 
    sqlx::query_as("SELECT id, at_ms, actor, action, target, detail FROM audit ORDER BY at_ms DESC, id DESC LIMIT ?")
693 
        .bind(limit)
694 
        .fetch_all(pool)
695 
        .await
696 
}
697
698 
/// The repositories an account can see beyond the public ones: its own,
699 
/// its organisations', and those it collaborates on.
700 
pub async fn visible_repo_ids(pool: &Pool, account_id: i64) -> sqlx::Result<Vec<i64>> {
701 
    sqlx::query_scalar(
702 
        "SELECT r.id FROM repositories r WHERE r.owner IN (SELECT username FROM accounts WHERE id = ?) \
703 
         OR r.owner IN (SELECT o.name FROM organisations o JOIN org_members m ON m.org_id = o.id WHERE m.account_id = ?) \
704 
         OR r.id IN (SELECT repo_id FROM collaborators WHERE account_id = ? OR team_id IN (SELECT team_id FROM team_members WHERE account_id = ?))",
705 
    )
706 
    .bind(account_id)
707 
    .bind(account_id)
708 
    .bind(account_id)
709 
    .bind(account_id)
710 
    .fetch_all(pool)
711 
    .await
712 
}
713
714 
// ------------------------------------------------------ account links
715
716 
/// A verification or reset link (hub spec §5.1.1).
717 
#[derive(Clone, Debug, FromRow)]
718 
pub struct AccountLink {
719 
    pub id: i64,
720 
    pub account_id: i64,
721 
    pub kind: String,
722 
    pub hash: String,
723 
    pub email: String,
724 
    pub created_by: Option<i64>,
725 
    pub created_ms: i64,
726 
    pub expires_ms: i64,
727 
    pub used_ms: Option<i64>,
728 
}
729
730 
/// Make a link; returns its secret (shown or sent once, stored as a hash).
731 
pub async fn create_link(
732 
    pool: &Pool,
733 
    account_id: i64,
734 
    kind: &str,
735 
    email: &str,
736 
    created_by: Option<i64>,
737 
    ttl_ms: i64,
738 
) -> sqlx::Result<String> {
739 
    let secret = crate::auth::random_hex(32);
740 
    let now = now_ms();
741 
    sqlx::query(
742 
        "INSERT INTO account_links (account_id, kind, hash, email, created_by, created_ms, expires_ms) VALUES (?, ?, ?, ?, ?, ?, ?)",
743 
    )
744 
    .bind(account_id)
745 
    .bind(kind)
746 
    .bind(crate::auth::token_hash(&secret))
747 
    .bind(email)
748 
    .bind(created_by)
749 
    .bind(now)
750 
    .bind(now + ttl_ms)
751 
    .execute(pool)
752 
    .await?;
753 
    Ok(secret)
754 
}
755
756 
/// An unused, unexpired link of this kind with this secret.
757 
pub async fn live_link(pool: &Pool, kind: &str, secret: &str) -> sqlx::Result<Option<AccountLink>> {
758 
    sqlx::query_as("SELECT * FROM account_links WHERE kind = ? AND hash = ? AND used_ms IS NULL AND expires_ms > ?")
759 
        .bind(kind)
760 
        .bind(crate::auth::token_hash(secret))
761 
        .bind(now_ms())
762 
        .fetch_optional(pool)
763 
        .await
764 
}
765
766 
/// Consume a link; false when it was already used (a race lost).
767 
pub async fn use_link(pool: &Pool, id: i64) -> sqlx::Result<bool> {
768 
    let r =
769 
        sqlx::query("UPDATE account_links SET used_ms = ? WHERE id = ? AND used_ms IS NULL").bind(now_ms()).bind(id).execute(pool).await?;
770 
    Ok(r.rows_affected() == 1)
771 
}
772
773 
/// Void every unused link of a kind for an account.
774 
pub async fn void_links(pool: &Pool, account_id: i64, kind: &str) -> sqlx::Result<()> {
775 
    sqlx::query("UPDATE account_links SET used_ms = ? WHERE account_id = ? AND kind = ? AND used_ms IS NULL")
776 
        .bind(now_ms())
777 
        .bind(account_id)
778 
        .bind(kind)
779 
        .execute(pool)
780 
        .await?;
781 
    Ok(())
782 
}
783
784 
/// When the account's newest link of a kind (for this address, when given) was made, for rate limits.
785 
pub async fn last_link_ms(pool: &Pool, account_id: i64, kind: &str, email: Option<&str>) -> sqlx::Result<Option<i64>> {
786 
    match email {
787 
        Some(e) => {
788 
            sqlx::query_scalar("SELECT MAX(created_ms) FROM account_links WHERE account_id = ? AND kind = ? AND email = ?")
789 
                .bind(account_id)
790 
                .bind(kind)
791 
                .bind(e)
792 
                .fetch_one(pool)
793 
                .await
794 
        }
795 
        None => {
796 
            sqlx::query_scalar("SELECT MAX(created_ms) FROM account_links WHERE account_id = ? AND kind = ?")
797 
                .bind(account_id)
798 
                .bind(kind)
799 
                .fetch_one(pool)
800 
                .await
801 
        }
802 
    }
803 
}
804
805 
/// How many links of a kind the account got since `since_ms`.
806 
pub async fn links_since(pool: &Pool, account_id: i64, kind: &str, since_ms: i64) -> sqlx::Result<i64> {
807 
    sqlx::query_scalar("SELECT COUNT(*) FROM account_links WHERE account_id = ? AND kind = ? AND created_ms >= ?")
808 
        .bind(account_id)
809 
        .bind(kind)
810 
        .bind(since_ms)
811 
        .fetch_one(pool)
812 
        .await
813 
}
814
815 
pub async fn set_email_verified(pool: &Pool, id: i64, verified: bool) -> sqlx::Result<()> {
816 
    sqlx::query("UPDATE accounts SET email_verified = ? WHERE id = ?").bind(verified).bind(id).execute(pool).await?;
817 
    Ok(())
818 
}
819
820 
/// The account with this verified email (case-insensitive), if exactly one.
821 
pub async fn account_by_verified_email(pool: &Pool, email: &str) -> sqlx::Result<Option<Account>> {
822 
    let found: Vec<Account> = sqlx::query_as("SELECT * FROM accounts WHERE email_verified = 1 AND lower(email) = lower(?) LIMIT 2")
823 
        .bind(email)
824 
        .fetch_all(pool)
825 
        .await?;
826 
    Ok(if found.len() == 1 { found.into_iter().next() } else { None })
827 
}
828
829 
/// Set or clear the picture of an account (`account`) or organisation (`org`).
830 
pub async fn set_avatar(pool: &Pool, kind: &str, id: i64, value: &str) -> sqlx::Result<()> {
831 
    let table = if kind == "org" { "organisations" } else { "accounts" };
832 
    sqlx::query(&format!("UPDATE {table} SET avatar = ? WHERE id = ?")).bind(value).bind(id).execute(pool).await?;
833 
    Ok(())
834 
}