commit d202e90d861bddc20685e6eb9725513109459ede · gmstn.systems/shard

+40 -1

src/index.ts

···

       1
       1
        
       import { setupDbWithMigrations } from "@/db";

     

       2
       2
       -
       import { DB_URL, SERVER_PORT } from "@/lib/env";

     

       2
       2
       +
       import { DB_URL, OWNER_DID, SERVER_PORT, SERVICE_DID } from "@/lib/env";

     

       3
       3
       +
       import { setRegistrationState } from "@/lib/state";

     

       4
       4
       +
       import type { AtUri } from "@/lib/types/atproto";

     

       5
       5
       +
       import { getRecordFromAtUri } from "@/lib/utils/atproto";

     

       3
       6
        
       import { newErrorResponse } from "@/lib/utils/http/responses";

     

       7
       7
       +
       import { connectToPrism } from "@/lib/utils/prism";

     

       4
       8
        
       import {

     

       9
       9
       +
           attachShardRegistrationListener,

     

       5
       10
        
           wrapHttpRegistrationCheck,

     

       6
       11
        
           wrapWsRegistrationCheck,

     

       7
       12
        
       } from "@/lib/utils/registration";

     
···

       60
       65
        
               server.log.error(err);

     

       61
       66
        
               process.exit(1);

     

       62
       67
        
           });

     

       68
       68
       +
       

     

       69
       69
       +
           let shardUrlOrigin = decodeURIComponent(

     

       70
       70
       +
               SERVICE_DID.startsWith("did:web:") ? SERVICE_DID.slice(8) : "",

     

       71
       71
       +
           );

     

       72
       72
       +
           if (shardUrlOrigin === "localhost")

     

       73
       73
       +
               shardUrlOrigin += `:${SERVER_PORT.toString()}`;

     

       74
       74
       +
           if (shardUrlOrigin === "") {

     

       75
       75
       +
               // TODO: resolve did:plc endpoint to get the origin of the shard endpoint described by the did:plc doc

     

       76
       76
       +
               // for now we just throw.

     

       77
       77
       +
               throw new Error(

     

       78
       78
       +
                   "did:plc support not yet implemented. Provide a did:web for now. did:plc support will come in the future.",

     

       79
       79
       +
               );

     

       80
       80
       +
           }

     

       81
       81
       +
       

     

       82
       82
       +
           const shardAtUri: Required<AtUri> = {

     

       83
       83
       +
               // @ts-expect-error alas, template literal weirdness continues uwu

     

       84
       84
       +
               authority: OWNER_DID,

     

       85
       85
       +
               collection: "systems.gmstn.development.shard",

     

       86
       86
       +
               rKey: shardUrlOrigin,

     

       87
       87
       +
           };

     

       88
       88
       +
       

     

       89
       89
       +
           const shardRecord = await getRecordFromAtUri(shardAtUri);

     

       90
       90
       +
       

     

       91
       91
       +
           if (shardRecord.ok) {

     

       92
       92
       +
               setRegistrationState(true);

     

       93
       93
       +
           }

     

       94
       94
       +
       

     

       95
       95
       +
           const prismWebsocket = connectToPrism({

     

       96
       96
       +
               wantedCollections: ["systems.gmstn.development.*"],

     

       97
       97
       +
           });

     

       98
       98
       +
       

     

       99
       99
       +
           // TODO: probably move this to an `attachListeners` hook that attaches the listeners we want.

     

       100
       100
       +
           // least tested. will probably have nuances we need to work on in the future

     

       101
       101
       +
           attachShardRegistrationListener(prismWebsocket);

     

       63
       102
        
       };

     

       64
       103
        
       

     

       65
       104
        
       main()

+24

src/lib/utils/prism.ts

···

       1
       1
       +
       // Prism is a jetstream/jetstream-compatible fork for receiving filtered events from the firehose

     

       2
       2
       +
       import { PRISM_URL } from "@/lib/env";

     

       3
       3
       +
       import WebSocket from "ws";

     

       4
       4
       +
       

     

       5
       5
       +
       export const connectToPrism = (opts?: {

     

       6
       6
       +
           wantedCollections?: Array<string>;

     

       7
       7
       +
           wantedDids?: Array<string>;

     

       8
       8
       +
           cursor?: number;

     

       9
       9
       +
       }) => {

     

       10
       10
       +
           const endpoint = PRISM_URL;

     

       11
       11
       +
           if (opts) {

     

       12
       12
       +
               const { wantedCollections, wantedDids, cursor } = opts;

     

       13
       13
       +
               if (wantedCollections)

     

       14
       14
       +
                   wantedCollections.forEach((collection) => {

     

       15
       15
       +
                       endpoint.searchParams.append("wantedCollections", collection);

     

       16
       16
       +
                   });

     

       17
       17
       +
               if (wantedDids)

     

       18
       18
       +
                   wantedDids.forEach((did) => {

     

       19
       19
       +
                       endpoint.searchParams.append("wantedDids", did);

     

       20
       20
       +
                   });

     

       21
       21
       +
               if (cursor) endpoint.searchParams.append("cursor", cursor.toString());

     

       22
       22
       +
           }

     

       23
       23
       +
           return new WebSocket(endpoint);

     

       24
       24
       +
       };

+5 -8

src/lib/utils/registration.ts

···

       15
       15
        
               if (!registrationState.registered) {

     

       16
       16
        
                   return newErrorResponse(503, {

     

       17
       17
        
                       message:

     

       18
       18
       -
                           "Lattice has not been registered for use. Register it in the dashboard or make the record yourself using the bootstrapper if you're doing local development.",

     

       18
       18
       +
                           "Shard has not been registered for use. Register it in the dashboard or make the record yourself using the bootstrapper if you're doing local development.",

     

       19
       19
        
                   });

     

       20
       20
        
               }

     

       21
       21
        
       

     
···

       31
       31
        
           const registrationState = getRegistrationState();

     

       32
       32
        
           const wrappedFunction: WsRouteHandler = (socket, request) => {

     

       33
       33
        
               if (!registrationState.registered) {

     

       34
       34
       -
                   socket.close(

     

       35
       35
       -
                       1013,

     

       36
       36
       -
                       "Service unavailable: Lattice not yet registered",

     

       37
       37
       -
                   );

     

       34
       34
       +
                   socket.close(1013, "Service unavailable: Shard not yet registered");

     

       38
       35
        
                   return;

     

       39
       36
        
               }

     

       40
       37
        
       

     
···

       44
       41
        
           return wrappedFunction;

     

       45
       42
        
       }

     

       46
       43
        
       

     

       47
       47
       -
       export const attachLatticeRegistrationListener = (socket: WebSocket) => {

     

       44
       44
       +
       export const attachShardRegistrationListener = (socket: WebSocket) => {

     

       48
       45
        
           socket.on("message", (rawData: RawData) => {

     

       49
       46
        
               const data = rawDataToString(rawData);

     

       50
       47
        
               const jsonData: unknown = JSON.parse(data);

     
···

       61
       58
        
               // TODO: replace empty string with call to resolve did doc and the endpoint and yadda yadda etc. etc. you get it.

     

       62
       59
        
               // if you don't, then the tl;dr is you need to resolve the did:plc document to get the service endpoint describing this lattice and ensure

     

       63
       60
        
               // that the domain/origin/whatever matches with the rkey (or record value if we decide to transition to that)

     

       64
       64
       -
               const latticeDomain = SERVICE_DID.startsWith("did:web:")

     

       61
       61
       +
               const shardDomain = SERVICE_DID.startsWith("did:web:")

     

       65
       62
        
                   ? SERVICE_DID.slice(8)

     

       66
       63
        
                   : "";

     

       67
       67
       -
               if (rkey !== latticeDomain) return;

     

       64
       64
       +
               if (rkey !== shardDomain) return;

     

       68
       65
        
       

     

       69
       66
        
               setRegistrationState(true);

     

       70
       67
        
           });