1+ import { EventHandler } from "sst/node/event-bus" ;
2+ import { extractRepositoryEvent } from "./events" ;
3+ import { Clerk } from "@clerk/clerk-sdk-node" ;
4+ import { createClient } from "@libsql/client" ;
5+ import { drizzle } from "drizzle-orm/libsql" ;
6+ import { getMembers } from "@acme/extract-functions" ;
7+ import type { Context , GetMembersEntities , GetMembersSourceControl } from "@acme/extract-functions" ;
8+ import { members , repositoriesToMembers } from "@acme/extract-schema" ;
9+ import type { Namespace , Repository } from "@acme/extract-schema" ;
10+ import { GitHubSourceControl , GitlabSourceControl } from "@acme/source-control" ;
11+ import type { Pagination } from "@acme/source-control" ;
12+ import { Config } from "sst/node/config" ;
13+ import { extractMemberPageBatchMessage } from "./messages" ;
14+
15+ import { QueueHandler } from "./create-message" ;
16+
17+ const clerkClient = Clerk ( { secretKey : Config . CLERK_SECRET_KEY } ) ;
18+ const client = createClient ( { url : Config . DATABASE_URL , authToken : Config . DATABASE_AUTH_TOKEN } ) ;
19+
20+ const fetchSourceControlAccessToken = async ( userId : string , forgeryIdProvider : 'oauth_github' | 'oauth_gitlab' ) => {
21+ const [ userOauthAccessTokenPayload , ...rest ] = await clerkClient . users . getUserOauthAccessToken ( userId , forgeryIdProvider ) ;
22+ if ( ! userOauthAccessTokenPayload ) throw new Error ( "Failed to get token" ) ;
23+ if ( rest . length !== 0 ) throw new Error ( "wtf ?" ) ;
24+
25+ return userOauthAccessTokenPayload . token ;
26+ }
27+
28+ const initSourceControl = async ( userId : string , sourceControl : 'github' | 'gitlab' ) => {
29+ const accessToken = await fetchSourceControlAccessToken ( userId , `oauth_${ sourceControl } ) ;
30+ if ( sourceControl === 'github' ) return new GitHubSourceControl ( accessToken ) ;
31+ if ( sourceControl === 'gitlab' ) return new GitlabSourceControl ( accessToken ) ;
32+ return null ;
33+ }
34+
35+ const db = drizzle ( client ) ;
36+
37+ const context : Context < GetMembersSourceControl , GetMembersEntities > = {
38+ entities : {
39+ members,
40+ repositoriesToMembers
41+ } ,
42+ integrations : {
43+ sourceControl : null ,
44+ } ,
45+ db,
46+ } ;
47+
48+ type ExtractMembersPageInput = {
49+ namespace : Namespace | null ;
50+ repository : Repository ;
51+ sourceControl : "github" | "gitlab" ;
52+ userId : string ;
53+ paginationInfo : Pagination | null ;
54+ }
55+
56+ const extractMembersPage = async ( { namespace, repository, sourceControl, userId, paginationInfo } : ExtractMembersPageInput ) => {
57+ const page = paginationInfo ?. page ;
58+ const perPage = paginationInfo ?. perPage ;
59+
60+ try {
61+ context . integrations . sourceControl = await initSourceControl ( userId , sourceControl ) ;
62+ } catch ( error ) {
63+ console . error ( error ) ;
64+ return ;
65+ }
66+
67+ console . log ( 'processing page' , paginationInfo ?. page , 'perPage' , paginationInfo ?. perPage ) ;
68+
69+ const { paginationInfo : resultPaginationInfo } = await getMembers ( {
70+ externalRepositoryId : repository . externalId ,
71+ namespaceName : namespace ?. name || "" ,
72+ repositoryId : repository . id ,
73+ repositoryName : repository . name ,
74+ perPage : page ,
75+ page : perPage
76+ } , context ) ;
77+
78+ return resultPaginationInfo ;
79+ } ;
80+
81+ const range = ( a : number , b : number ) => Array . apply ( 0 , { length : b - a + 1 } as number [ ] ) . map ( ( _ , index ) => index + a ) ;
82+ const chunks = < T > ( array : Array < T > , size : number ) => Array . apply ( 0 , { length : Math . ceil ( array . length / size ) } as unknown [ ] ) . map ( ( _ , index ) => array . slice ( index * size , ( index + 1 ) * size ) ) ;
83+
84+ export const eventHandler = EventHandler ( extractRepositoryEvent , async ( ev ) => {
85+
86+ const pagination = await extractMembersPage ( {
87+ namespace : ev . properties . namespace ,
88+ repository : ev . properties . repository ,
89+ sourceControl : ev . metadata . sourceControl ,
90+ userId : ev . metadata . userId ,
91+ paginationInfo : { page : 1 , perPage : 2 , totalPages : 1000 } ,
92+ } ) ;
93+
94+ if ( ! pagination ) return ;
95+
96+ const remainingMemberPages = range ( 2 , pagination . totalPages )
97+ . map ( page => ( {
98+ page,
99+ perPage : pagination . perPage ,
100+ totalPages : pagination . totalPages
101+ } satisfies Pagination ) ) ;
102+
103+ const batchedPages = chunks ( remainingMemberPages , 10 ) ;
104+
105+ await Promise . all ( batchedPages . map ( batch => extractMemberPageBatchMessage . send (
106+ batch . map ( page => ( {
107+ namespace : ev . properties . namespace ,
108+ repository : ev . properties . repository ,
109+ pagination : page
110+ } ) ) , {
111+ version : 1 ,
112+ caller : 'extract-member' ,
113+ sourceControl : ev . metadata . sourceControl ,
114+ userId : ev . metadata . userId ,
115+ timestamp : new Date ( ) . getTime ( ) ,
116+ } ) ) ) ;
117+ } ) ;
118+
119+ export const queueHandler = QueueHandler ( extractMemberPageBatchMessage , async ( messages ) => {
120+ console . log ( 'CONSUMER BARRIER' , messages . length )
121+ // TODO: partial retries ?
122+ await Promise . all ( messages . map ( message => extractMembersPage ( {
123+ namespace : message . content . namespace ,
124+ paginationInfo : message . content . pagination ,
125+ repository : message . content . repository ,
126+ sourceControl : message . metadata . sourceControl ,
127+ userId : message . metadata . userId
128+ } ) ) )
129+ } )
0 commit comments