Differentiate pegasus logging from dream logging

pegasus/lib/api/account_/migrate/migrate.ml

···
           | Ok () ->
               (true, None)
           | Error e ->
-              Dream.warning (fun log ->
+              Log.warn (fun log ->
                   log "migration %s: failed to deactivate old account: %s" did e ) ;
               (false, Some e)
         in
···
     | Some session -> (
       match%lwt Remote.request_plc_signature old_client with
       | Error e ->
-          Dream.warning (fun log ->
+          Log.warn (fun log ->
               log "migration %s: failed to request PLC signature: %s" did e ) ;
           let%lwt () =
             State.set ctx.req
···
         | Ok () ->
             (true, None)
         | Error err ->
-            Dream.warning (fun log ->
+            Log.warn (fun log ->
                 log "migration %s: failed to deactivate old account: %s" did err ) ;
             (false, Some err)
       in
···
         let%lwt () =
           match%lwt Ops.check_account_status ~did with
           | Ok status ->
-              Dream.info (fun log ->
+              Log.info (fun log ->
                   log
                     "migration %s: repo imported, indexed_records=%d, \
                      expected_blobs=%d"
                     did status.indexed_records status.expected_blobs ) ;
               Lwt.return_unit
           | Error e ->
-              Dream.warning (fun log ->
+              Log.warn (fun log ->
                   log "migration %s: failed to check account status: %s" did e ) ;
               Lwt.return_unit
         in
         match%lwt Ops.list_missing_blobs ~did ~limit:50 () with
         | Error e ->
-            Dream.warning (fun log ->
+            Log.warn (fun log ->
                 log "migration %s: failed to list missing blobs: %s" did e ) ;
             transition_to_plc_token_step ctx ~old_client ~old_pds ~did ~handle
               ~email ~blobs_imported:0 ~blobs_failed:0
···
           Ops.list_missing_blobs ~did:state.did ~limit:50 ?cursor ()
         with
         | Error e ->
-            Dream.warning (fun log ->
+            Log.warn (fun log ->
                 log "migration %s: failed to list missing blobs: %s" state.did e ) ;
             transition_to_plc_token_step ctx ~old_client ~old_pds:state.old_pds
               ~did:state.did ~handle:state.handle ~email:state.email
···
               | Ok _ ->
                   Lwt.return []
               | Error e ->
-                  Dream.warning (fun log ->
+                  Log.warn (fun log ->
                       log "migration %s: failed to get old PDS credentials: %s"
                         state.did e ) ;
                   Lwt.return []
···
                       let%lwt () =
                         match%lwt Ops.check_account_status ~did:state.did with
                         | Ok status ->
-                            Dream.info (fun log ->
+                            Log.info (fun log ->
                                 log
                                   "migration %s: activating account, \
                                    imported_blobs=%d/%d"
···
                                   status.expected_blobs ) ;
                             Lwt.return_unit
                         | Error e ->
-                            Dream.warning (fun log ->
+                            Log.warn (fun log ->
                                 log
                                   "migration %s: failed to check status before \
                                    activation: %s"
···
                         | Ok () ->
                             (true, None)
                         | Error e ->
-                            Dream.warning (fun log ->
+                            Log.warn (fun log ->
                                 log
                                   "migration %s: failed to deactivate old \
                                    account: %s"
···
                   | Ok () ->
                       (true, None)
                   | Error e ->
-                      Dream.warning (fun log ->
+                      Log.warn (fun log ->
                           log
                             "migration %s: failed to deactivate old account: %s"
                             did e ) ;
···
                 | Ok () -> (
                   match%lwt Ops.list_missing_blobs ~did ~limit:50 () with
                   | Error e ->
-                      Dream.warning (fun log ->
+                      Log.warn (fun log ->
                           log "migration %s: failed to list missing blobs: %s"
                             did e ) ;
                       transition_to_plc_token_step ctx ~old_client:client

pegasus/lib/api/account_/migrate/ops.ml

···
       (fun cid_str ->
         match%lwt Remote.fetch_blob ~did ~cid:cid_str client with
         | Error e ->
-            Dream.warning (fun log ->
+            Log.warn (fun log ->
                 log "migration %s: failed to fetch blob %s: %s" did cid_str e ) ;
             Lwt.return_error cid_str
         | Ok (data, mimetype) -> (

pegasus/lib/api/account_/migrate/remote.ml

···
     Lwt.return_ok res.preferences
   with
   | Hermes.Xrpc_error {status; error; _} ->
-      Dream.warning (fun log ->
+      Log.warn (fun log ->
           log "migration: failed to fetch preferences: %d %s" status error ) ;
       Lwt.return_ok []
   | exn ->
-      Dream.warning (fun log ->
+      Log.warn (fun log ->
           log "migration: exception fetching preferences: %s"
             (Printexc.to_string exn) ) ;
       Lwt.return_ok []

pegasus/lib/api/identity/resolveHandle.ml

···
         | Error e -> (
           try%lwt Xrpc.service_proxy ctx
           with _ ->
-            Dream.error (fun log -> log "%s" e) ;
+            Log.err (fun log -> log "%s" e) ;
             Errors.internal_error ~msg:"could not resolve handle" () ) ) )

pegasus/lib/api/identity/updateHandle.ml

···
           Dream.empty `OK
       | Error e ->
           let msg = update_handle_error_to_string e in
-          Dream.error (fun log -> log ~request:req "%s" msg) ;
+          Log.err (fun log -> log "%s" msg) ;
           Errors.invalid_request ~name:"InvalidHandle" msg )

pegasus/lib/api/oauth_/par.ml

···
       let%lwt client =
         try%lwt Client.fetch_client_metadata req.client_id
         with e ->
-          Errors.log_exn ~req:ctx.req e ;
+          Errors.log_exn e ;
           Errors.invalid_request "failed to fetch client metadata"
       in
       if req.response_type <> "code" then

pegasus/lib/api/proxy/appBskyFeedGetFeed.ml

···
                 | None ->
                     Errors.invalid_request "missing proxy header" )
             with e ->
-              Dream.error (fun log ->
+              Log.err (fun log ->
                   log "failed to fetch feed generator record: %s"
                     (Printexc.to_string e) ) ;
               Errors.internal_error ~msg:"failed to fetch feed generator record"

pegasus/lib/api/server/requestPasswordReset.ml

···
             | Some actor ->
                 Lwt.return_some actor
             | None ->
-                Dream.log "password reset requested for unknown email: %s" email ;
                 Lwt.return_none )
       in
       match actor_opt with

pegasus/lib/api/server/resetPassword.ml

···
     (fun {req; db; _} ->
       let%lwt {token; password} = Xrpc.parse_body req input_of_yojson in
       match%lwt reset_password ~token ~password db with
-      | Ok did ->
-          Dream.log "password reset completed for %s" did ;
+      | Ok _ ->
           Dream.empty `OK
       | Error InvalidToken ->
           Errors.invalid_request ~name:"InvalidToken" "invalid or expired token"

pegasus/lib/auth.ml

···
       let now_s = int_of_float (Unix.gettimeofday ()) in
       match Jwt.symmetric_jwt_of_yojson payload with
       | Error e ->
-          Dream.debug (fun log -> log "bearer jwt decode error: %s" e) ;
+          Log.debug (fun log -> log "bearer jwt decode error: %s" e) ;
           Lwt.return_error "invalid token format"
       | Ok jwt ->
           if jwt.aud <> Env.did then Lwt.return_error "invalid aud"
···
     | Error "use_dpop_nonce" ->
         Lwt.return_error @@ Errors.use_dpop_nonce ()
     | Error e ->
-        Dream.debug (fun log -> log ~request:req "dpop error: %s" e) ;
+        Log.debug (fun log -> log "dpop error: %s" e) ;
         Lwt.return_error @@ Errors.invalid_request ("dpop error: " ^ e)
     | Ok proof ->
         Lwt.return_ok (DPoP {proof})
···
    fun {req; db} ->
     match parse_dpop req with
     | Error e ->
-        Dream.debug (fun log -> log ~request:req "dpop error: %s" e) ;
+        Log.debug (fun log -> log "dpop error: %s" e) ;
         Lwt.return_error @@ Errors.invalid_request ("dpop error: " ^ e)
     | Ok token -> (
       match
···
       | Error "use_dpop_nonce" ->
           Lwt.return_error @@ Errors.use_dpop_nonce ()
       | Error e ->
-          Dream.debug (fun log -> log ~request:req "dpop error: %s" e) ;
+          Log.debug (fun log -> log "dpop error: %s" e) ;
           Lwt.return_error @@ Errors.invalid_request ("dpop error: " ^ e)
       | Ok proof -> (
         match Jwt.verify_jwt token ~pubkey:Env.jwt_pubkey with
         | Error e ->
-            Dream.debug (fun log -> log ~request:req "invalid jwt: %s" e) ;
+            Log.debug (fun log -> log "invalid jwt: %s" e) ;
             Lwt.return_error @@ Errors.auth_required e
         | Ok (_header, claims) -> (
             let open Yojson.Safe.Util in
···
       | Ok _ ->
           check_actor_status did db
       | Error e ->
-          Dream.debug (fun log -> log "service jwt verification failed: %s" e) ;
+          Log.debug (fun log -> log "service jwt verification failed: %s" e) ;
           Lwt.return_error
           @@ Errors.invalid_request ~name:"InvalidToken"
                "jwt signature does not match jwt issuer"

pegasus/lib/blob_store.ml

···
       | Ok () ->
           Lwt.return_unit
       | Error e ->
-          Dream.error (fun log ->
+          Log.err (fun log ->
               log "S3 delete failed for %s: %s" key (Util.s3_error_to_string e) ) ;
           Lwt.return_unit )
   | None ->

pegasus/lib/env.ml

···
   match getenv_opt "PDS_DPOP_NONCE_SECRET" ~default:"" with
   | "" ->
       let secret = Mirage_crypto_rng_unix.getrandom 32 in
-      Dream.warning (fun log ->
+      Log.warn (fun log ->
           log "PDS_DPOP_NONCE_SECRET not set; using PDS_DPOP_NONCE_SECRET=%s"
             ( Base64.(encode ~alphabet:uri_safe_alphabet ~pad:false) secret
             |> Result.get_ok ) ) ;

pegasus/lib/errors.ml

···
   in
   match exn with
   | InvalidRequestError (error, message) ->
-      Dream.debug (fun log -> log "invalid request: %s - %s" error message) ;
+      Log.debug (fun log -> log "invalid request: %s - %s" error message) ;
       format_response error message `Bad_Request
   | InternalServerError (error, message) ->
-      Dream.debug (fun log ->
-          log "internal server error: %s - %s" error message ) ;
+      Log.debug (fun log -> log "internal server error: %s - %s" error message) ;
       format_response error message `Internal_Server_Error
   | AuthError (error, message) ->
-      Dream.debug (fun log -> log "auth error: %s - %s" error message) ;
+      Log.debug (fun log -> log "auth error: %s - %s" error message) ;
       format_response error message `Unauthorized
   | NotFoundError (error, message) ->
-      Dream.debug (fun log -> log "not found error: %s - %s" error message) ;
+      Log.debug (fun log -> log "not found error: %s - %s" error message) ;
       format_response error message `Not_Found
   | UseDpopNonceError ->
-      Dream.debug (fun log -> log "use_dpop_nonce error") ;
+      Log.debug (fun log -> log "use_dpop_nonce error") ;
       Dream.json ~status:`Bad_Request
         ~headers:
           [ ("WWW-Authenticate", {|DPoP error="use_dpop_nonce"|})
           ; ("Access-Control-Expose-Headers", "DPoP-Nonce, WWW-Authenticate") ]
         {|{ "error": "use_dpop_nonce" }|}
   | e ->
-      Dream.warning (fun log ->
+      Log.warn (fun log ->
           log "unexpected internal error: %s" (Printexc.to_string e) ) ;
       format_response "InternalServerError" "Internal server error"
         `Internal_Server_Error
 
-let log_exn ?req exn =
-  Dream.error (fun log -> log ?request:req "%s" (Printexc.to_string exn))
+let log_exn = Log.log_exn

pegasus/lib/jwt.ml

···
               let did = Option.value did ~default:iss_did in
               match%lwt Id_resolver.Did.resolve did with
               | Error e ->
-                  Dream.debug (fun log ->
+                  Log.debug (fun log ->
                       log "failed to resolve did %s: %s" did e ) ;
                   Lwt.return_error
                   @@ InternalError "could not resolve jwt issuer did"

pegasus/lib/log.ml

···
+let default_src = Logs.Src.create "pegasus"
+
+let debug ?(src = default_src) = Logs.debug ~src
+
+let err ?(src = default_src) = Logs.err ~src
+
+let warn ?(src = default_src) = Logs.warn ~src
+
+let info ?(src = default_src) = Logs.info ~src
+
+let log_exn exn = err (fun log -> log "%s" (Printexc.to_string exn))

pegasus/lib/oauth/dpop.ml

···
             else
               match header |> member "jwk" |> ec_jwk_of_yojson with
               | Error e ->
-                  Dream.debug (fun log -> log "error parsing jwk: %s" e) ;
+                  Log.debug (fun log -> log "error parsing jwk: %s" e) ;
                   Errors.internal_error ()
               | Ok jwk -> (
                   if

pegasus/lib/s3/backup.ml

···
   in
   match result with
   | Ok _ ->
-      Dream.info (fun log -> log "S3 backup uploaded: %s" key) ;
+      Log.info (fun log -> log "S3 backup uploaded: %s" key) ;
       Lwt.return_unit
   | Error e ->
-      Dream.error (fun log ->
+      Log.err (fun log ->
           log "S3 backup upload failed for %s: %s" key
             (Util.s3_error_to_string e) ) ;
       Lwt.return_unit
···
   if Sys.file_exists db_path then (
     let timestamp = timestamp_string () in
     let key = Printf.sprintf "backups/pegasus-%s.db" timestamp in
-    Dream.info (fun log -> log "starting main database backup") ;
+    Log.info (fun log -> log "starting main database backup") ;
     upload_to_s3 ~config ~file_path:db_path ~key )
   else (
-    Dream.warning (fun log -> log "main database not found: %s" db_path) ;
+    Log.warn (fun log -> log "main database not found: %s" db_path) ;
     Lwt.return_unit )
 
 let backup_user_db ~config ~did : unit Lwt.t =
···
   else Lwt.return_unit
 
 let backup_all_user_dbs ~config ~ds : unit Lwt.t =
-  Dream.info (fun log -> log "starting backup of user databases") ;
+  Log.info (fun log -> log "starting backup of user databases") ;
   let rec backup_batch cursor count =
     let%lwt actors = Data_store.list_actors ~cursor ~limit:100 ds in
     match actors with
     | [] ->
-        Dream.info (fun log -> log "backed up %d user databases" count) ;
+        Log.info (fun log -> log "backed up %d user databases" count) ;
         Lwt.return_unit
     | actors ->
         let%lwt () =
···
 let do_backup () : unit Lwt.t =
   match Env.s3_config with
   | Some config when config.backups_enabled ->
-      Dream.info (fun log -> log "starting S3 backup") ;
+      Log.info (fun log -> log "starting S3 backup") ;
       let%lwt ds = Data_store.connect () in
       let%lwt () = backup_main_db ~config in
       let%lwt () = backup_all_user_dbs ~config ~ds in
-      Dream.info (fun log -> log "S3 backup completed") ;
+      Log.info (fun log -> log "S3 backup completed") ;
       Lwt.return_unit
   | _ ->
       Lwt.return_unit
···
 let queue_backups () : unit Lwt.t =
   match Env.s3_config with
   | Some config when config.backups_enabled ->
-      Dream.info (fun log -> log "starting hourly S3 backups") ;
+      Log.info (fun log -> log "starting hourly S3 backups") ;
       let%lwt () =
         Lwt.catch
           (fun () -> do_backup ())
           (fun e ->
-            Dream.error (fun log ->
-                log "backup failed: %s" (Printexc.to_string e) ) ;
+            Log.err (fun log -> log "backup failed: %s" (Printexc.to_string e)) ;
             Lwt.return_unit )
       in
       let rec loop () =
···
           Lwt.catch
             (fun () -> do_backup ())
             (fun e ->
-              Dream.error (fun log ->
+              Log.err (fun log ->
                   log "backup failed: %s" (Printexc.to_string e) ) ;
               Lwt.return_unit )
         in

pegasus/lib/s3/blob_migration.ml

···
 let migrate_user ~did : (int * int) Lwt.t =
   match Env.s3_config with
   | Some config when config.blobs_enabled ->
-      Dream.info (fun log -> log "migrating blobs for user %s" did) ;
+      Log.info (fun log -> log "migrating blobs for user %s" did) ;
       let%lwt user_db = User_store.connect did in
       let migrated = ref 0 in
       let errors = ref 0 in
···
                         Sys.remove local_path ; incr migrated ; Lwt.return_unit
                         )
                       else (
-                        Dream.warning (fun log ->
+                        Log.warn (fun log ->
                             log "local blob file not found: %s" local_path ) ;
                         Lwt.return_unit ) )
                     (fun e ->
-                      Dream.error (fun log ->
+                      Log.err (fun log ->
                           log "blob migration error for %s: %s"
                             (Cid.to_string cid) (Printexc.to_string e) ) ;
                       incr errors ;
···
             migrate_batch (Cid.to_string last_cid)
       in
       let%lwt () = migrate_batch "" in
-      Dream.info (fun log ->
+      Log.info (fun log ->
           log "blob migration complete for %s: %d migrated, %d errors" did
             !migrated !errors ) ;
       Lwt.return (!migrated, !errors)
   | _ ->
-      Dream.error (fun log -> log "S3 blob storage not enabled") ;
+      Log.err (fun log -> log "S3 blob storage not enabled") ;
       Lwt.return (0, 0)
 
 let migrate_all () : unit Lwt.t =
   match Env.s3_config with
   | Some config when config.blobs_enabled ->
-      Dream.info (fun log -> log "migrating all blobs to S3") ;
+      Log.info (fun log -> log "migrating all blobs to S3") ;
       let%lwt ds = Data_store.connect () in
       let total_migrated = ref 0 in
       let total_errors = ref 0 in
···
             migrate_batch last.did
       in
       let%lwt () = migrate_batch "" in
-      Dream.info (fun log ->
+      Log.info (fun log ->
           log "blob migration complete: %d total migrated, %d total errors"
             !total_migrated !total_errors ) ;
       Lwt.return_unit
   | _ ->
-      Dream.error (fun log -> log "S3 blob storage not enabled") ;
+      Log.err (fun log -> log "S3 blob storage not enabled") ;
       Lwt.return_unit

pegasus/lib/sequencer.ml

···
                   Lexicons.([%xrpc post "com.atproto.sync.requestCrawl"])
                     ~hostname:Env.hostname client )
                 (fun exn ->
-                  Dream.warning (fun log ->
+                  Log.warn (fun log ->
                       log "failed to requestCrawl %s: %s" service
                         (Printexc.to_string exn) ) ) )
             Env.crawlers

pegasus/lib/util.ml

···
             recipients
           |> Option.get
         in
-        Dream.log "email to %s: %s" to_addr text
+        Log.info (fun log -> log "email to %s: %s" to_addr text)
   in
   match (Env.smtp_config, Env.smtp_sender) with
   | Some config, Some sender -> (
···
     | Ok message -> (
       try%lwt Letters.send ~config ~sender ~recipients ~message
       with e ->
-        Errors.log_exn e ;
+        Log.log_exn e ;
         Lwt.return (log_email ()) ) )
   | _ ->
       Lwt.return (log_email ())

pegasus/lib/xrpc.ml

···
             apply_rate_limits ~nsid rate_limits ctx ;
             try%lwt hdlr ctx
             with e ->
-              if not (is_xrpc_error e) then log_exn ~req:init.req e ;
+              if not (is_xrpc_error e) then log_exn e ;
               exn_to_response e
           with Rate_limiter.Rate_limit_exceeded status ->
             rate_limit_response status )
···
     | Rate_limiter.Rate_limit_exceeded status ->
         rate_limit_response status
     | e ->
-        if not (is_xrpc_error e) then log_exn ~req:init.req e ;
+        if not (is_xrpc_error e) then log_exn e ;
         exn_to_response e
   with Redirect r -> Dream.redirect init.req r
 
···
     in
     match query_json |> of_yojson with
     | Error e ->
-        Dream.debug (fun log -> log "error parsing query: %s" e) ;
+        Log.debug (fun log -> log "error parsing query: %s" e) ;
         Errors.internal_error ()
     | Ok query ->
         query
···
         | _ ->
             Lwt.return (`Assoc []) )
     in
+    Log.debug (fun l -> l "body: %s" (Yojson.Safe.to_string body_assoc)) ;
     match of_yojson body_assoc with
     | Error e ->
-        Dream.debug (fun log -> log "error parsing body: %s" e) ;
+        Log.debug (fun log -> log "error parsing body: %s" e) ;
         Errors.internal_error ()
     | Ok body ->
         Lwt.return body
···
                   Body.to_stream body |> Lwt_stream.iter_s (Dream.write stream) )
           | e ->
               let%lwt () = Body.drain_body body in
-              Dream.error (fun log ->
+              Log.err (fun log ->
                   log "error when proxying to %s: %s" (Uri.to_string uri)
                     (Http.Status.to_string e) ) ;
               Errors.internal_error ~msg:"failed to proxy request" () )
···
                   Body.to_stream body |> Lwt_stream.iter_s (Dream.write stream) )
           | e ->
               let%lwt () = Body.drain_body body in
-              Dream.error (fun log ->
+              Log.err (fun log ->
                   log "error when proxying to %s: %s" (Uri.to_string uri)
                     (Http.Status.to_string e) ) ;
               Errors.internal_error ~msg:"failed to proxy request" () )
       | _ ->
           Errors.invalid_request "unsupported method" )
   | Error e ->
-      Dream.error (fun log ->
-          log ~request:ctx.req "error when resolving destination service: %s" e ) ;
+      Log.err (fun log -> log "error when resolving destination service: %s" e) ;
       Errors.internal_error ~msg:"failed to resolve destination service" ()
 
 let service_proxy_handler db req =