objective categorical abstract machine language personal data server
Notify crawlers on event emit, add PDS_ prefix to env vars
+76
-26
+40
-26
pegasus/lib/env.ml
+40
-26
pegasus/lib/env.ml
···
2
2
try Sys.getenv name
3
3
with Not_found -> failwith ("Missing environment variable " ^ name)
4
4
5
-
let data_dir = Option.value ~default:"./data" @@ Sys.getenv_opt "DATA_DIR"
5
+
let data_dir = Option.value ~default:"./data" @@ Sys.getenv_opt "PDS_DATA_DIR"
6
6
7
7
let hostname = getenv "PDS_HOSTNAME"
8
8
···
11
11
let did =
12
12
Option.value ~default:("did:web:" ^ hostname) @@ Sys.getenv_opt "PDS_DID"
13
13
14
-
let invite_required = getenv "INVITE_CODE_REQUIRED" = "true"
14
+
let invite_required = getenv "PDS_INVITE_CODE_REQUIRED" = "true"
15
15
16
-
let rotation_key = getenv "ROTATION_KEY_MULTIBASE" |> Kleidos.parse_multikey_str
16
+
let rotation_key =
17
+
getenv "PDS_ROTATION_KEY_MULTIBASE" |> Kleidos.parse_multikey_str
17
18
18
-
let jwt_key = getenv "JWK_MULTIBASE" |> Kleidos.parse_multikey_str
19
+
let jwt_key = getenv "PDS_JWK_MULTIBASE" |> Kleidos.parse_multikey_str
19
20
20
21
let jwt_pubkey = Kleidos.derive_pubkey jwt_key
21
22
22
-
let admin_password = getenv "ADMIN_PASSWORD"
23
+
let admin_password = getenv "PDS_ADMIN_PASSWORD"
24
+
25
+
let crawlers =
26
+
Sys.getenv_opt "PDS_CRAWLERS"
27
+
|> Option.value ~default:"https://bsky.network"
28
+
|> String.split_on_char ','
29
+
|> List.map (fun u ->
30
+
match Uri.of_string @@ String.trim u with
31
+
| uri when Uri.host uri <> None ->
32
+
uri
33
+
| _ ->
34
+
Uri.make ~scheme:"https" ~host:u () )
23
35
24
36
let dpop_nonce_secret =
25
-
match Sys.getenv_opt "DPOP_NONCE_SECRET" with
37
+
match Sys.getenv_opt "PDS_DPOP_NONCE_SECRET" with
26
38
| Some sec ->
27
39
let secret =
28
40
Base64.(decode_exn ~alphabet:uri_safe_alphabet ~pad:false) sec
29
41
|> Bytes.of_string
30
42
in
31
43
if Bytes.length secret = 32 then secret
32
-
else failwith "DPOP_NONCE_SECRET must be 32 bytes in base64uri"
44
+
else failwith "PDS_DPOP_NONCE_SECRET must be 32 bytes in base64uri"
33
45
| None ->
34
46
let secret = Mirage_crypto_rng_unix.getrandom 32 in
35
47
Dream.warning (fun log ->
36
-
log "DPOP_NONCE_SECRET not set; using DPOP_NONCE_SECRET=%s"
48
+
log "PDS_DPOP_NONCE_SECRET not set; using PDS_DPOP_NONCE_SECRET=%s"
37
49
( Base64.(encode ~alphabet:uri_safe_alphabet ~pad:false) secret
38
50
|> Result.get_ok ) ) ;
39
51
Bytes.of_string secret
···
41
53
let smtp_config, smtp_sender =
42
54
begin
43
55
let with_starttls =
44
-
Option.value ~default:"false" @@ Sys.getenv_opt "SMTP_STARTTLS" = "true"
56
+
Option.value ~default:"false" @@ Sys.getenv_opt "PDS_SMTP_STARTTLS"
57
+
= "true"
45
58
in
46
59
match
47
-
( Option.map Uri.of_string (Sys.getenv_opt "SMTP_AUTH_URI")
48
-
, Sys.getenv_opt "SMTP_SENDER" )
60
+
( Option.map Uri.of_string (Sys.getenv_opt "PDS_SMTP_AUTH_URI")
61
+
, Sys.getenv_opt "PDS_SMTP_SENDER" )
49
62
with
50
63
| Some uri, Some sender -> (
51
64
match
···
66
79
, Some sender )
67
80
| Error _ ->
68
81
failwith
69
-
"SMTP_SENDER should be a valid mailbox, e.g. `e@mail.com` or \
82
+
"PDS_SMTP_SENDER should be a valid mailbox, e.g. `e@mail.com` or \
70
83
`Name <e@mail.com>`" )
71
84
| _ ->
72
85
failwith
73
-
"SMTP_AUTH_URI must be a valid smtp:// or smtps:// URI with \
86
+
"PDS_SMTP_AUTH_URI must be a valid smtp:// or smtps:// URI with \
74
87
username, password, and hostname" )
75
88
| Some _, None ->
76
89
failwith
77
-
"SMTP_SENDER must be set alongside SMTP_AUTH_URI; it should look \
78
-
like `e@mail.com` or `Name <e@mail.com>`"
90
+
"PDS_SMTP_SENDER must be set alongside PDS_SMTP_AUTH_URI; it should \
91
+
look like `e@mail.com` or `Name <e@mail.com>`"
79
92
| None, Some _ ->
80
-
failwith "SMTP_AUTH_URI must be set alongside SMTP_SENDER"
93
+
failwith "PDS_SMTP_AUTH_URI must be set alongside PDS_SMTP_SENDER"
81
94
| None, None ->
82
95
(None, None)
83
96
end
···
99
112
begin
100
113
let default_backup_interval = 3600.0 in
101
114
let blobs_enabled =
102
-
Sys.getenv_opt "S3_BLOBS_ENABLED" |> Option.map (( = ) "true")
115
+
Sys.getenv_opt "PDS_S3_BLOBS_ENABLED" |> Option.map (( = ) "true")
103
116
in
104
117
let backups_enabled =
105
-
Sys.getenv_opt "S3_BACKUPS_ENABLED" |> Option.map (( = ) "true")
118
+
Sys.getenv_opt "PDS_S3_BACKUPS_ENABLED" |> Option.map (( = ) "true")
106
119
in
107
120
let backup_interval_s =
108
-
Sys.getenv_opt "S3_BACKUP_INTERVAL_S"
121
+
Sys.getenv_opt "PDS_S3_BACKUP_INTERVAL_S"
109
122
|> Option.map float_of_string_opt
110
123
|> Option.join
111
124
in
112
-
let endpoint = Sys.getenv_opt "S3_ENDPOINT" in
125
+
let endpoint = Sys.getenv_opt "PDS_S3_ENDPOINT" in
113
126
match (blobs_enabled, backups_enabled) with
114
127
| Some true, _ | _, Some true -> (
115
128
match
116
-
( Sys.getenv_opt "S3_REGION"
117
-
, Sys.getenv_opt "S3_BUCKET"
118
-
, Sys.getenv_opt "S3_ACCESS_KEY"
119
-
, Sys.getenv_opt "S3_SECRET_KEY" )
129
+
( Sys.getenv_opt "PDS_S3_REGION"
130
+
, Sys.getenv_opt "PDS_S3_BUCKET"
131
+
, Sys.getenv_opt "PDS_S3_ACCESS_KEY"
132
+
, Sys.getenv_opt "PDS_S3_SECRET_KEY" )
120
133
with
121
134
| Some region, Some bucket, Some access_key, Some secret_key ->
122
135
let region_obj =
···
139
152
; bucket
140
153
; access_key
141
154
; secret_key
142
-
; cdn_url= Sys.getenv_opt "S3_CDN_URL"
155
+
; cdn_url= Sys.getenv_opt "PDS_S3_CDN_URL"
143
156
; credentials_obj=
144
157
Aws_s3.Credentials.make ~access_key ~secret_key ()
145
158
; endpoint_obj }
146
159
| _ ->
147
160
failwith
148
161
"S3 backups require the following environment variables: \
149
-
S3_REGION, S3_BUCKET, S3_ACCESS_KEY, S3_SECRET_KEY" )
162
+
PDS_S3_REGION, PDS_S3_BUCKET, PDS_S3_ACCESS_KEY, \
163
+
PDS_S3_SECRET_KEY" )
150
164
| _ ->
151
165
None
152
166
end
+36
pegasus/lib/sequencer.ml
+36
pegasus/lib/sequencer.ml
···
458
458
459
459
let queue_max = 1000
460
460
461
+
let notify_interval = 20 * Util.minute
462
+
461
463
let ring : item array = Array.make ring_size {seq= 0; bytes= Bytes.empty}
462
464
463
465
let head_seq = ref 0
···
477
479
478
480
let lock = Lwt_mutex.create ()
479
481
482
+
let last_notified = ref 0
483
+
480
484
let publish (it : item) =
481
485
Lwt_mutex.with_lock lock (fun () ->
482
486
head_seq := it.seq ;
483
487
ring.(it.seq mod ring_size) <- it ;
484
488
if !count < ring_size then incr count ;
489
+
let now = Util.now_ms () in
490
+
if now - !last_notified > notify_interval then begin
491
+
last_notified := now ;
492
+
List.iter
493
+
(fun crawler ->
494
+
let uri =
495
+
Uri.with_path crawler "/xrpc/com.atproto.sync.requestCrawl"
496
+
in
497
+
Lwt.dont_wait
498
+
(fun () ->
499
+
let%lwt res, _ =
500
+
Cohttp_lwt_unix.Client.post
501
+
~headers:
502
+
(Cohttp.Header.of_list
503
+
[("Content-Type", "application/json")] )
504
+
~body:
505
+
(Printf.ksprintf Cohttp_lwt.Body.of_string
506
+
{|{ "hostname": "%s" |} Env.hostname )
507
+
uri
508
+
in
509
+
match res.status with
510
+
| `OK ->
511
+
Lwt.return_unit
512
+
| status ->
513
+
failwith
514
+
("errored with status " ^ Http.Status.to_string status) )
515
+
(fun exn ->
516
+
Dream.warning (fun log ->
517
+
log "failed to requestCrawl %s: %s"
518
+
(Uri.to_string crawler) (Printexc.to_string exn) ) ) )
519
+
Env.crawlers
520
+
end ;
485
521
Hashtbl.iter
486
522
(fun _ s ->
487
523
if not s.closed then (