byarielm.fyi
ATlast — you'll never need to find your favorites on another platform again. Find your favs in the ATmosphere.
atproto
add bulk inserts to db, replace indiv queries
+290
-145
+201
-82
netlify/functions/db-helpers.ts
+201
-82
netlify/functions/db-helpers.ts
···
1
1
import { getDbClient } from './db';
2
2
3
-
// Normalize username for consistent matching
4
-
export function normalizeUsername(username: string): string {
5
-
return username.toLowerCase().replace(/[._-]/g, '');
3
+
export async function createUpload(
4
+
uploadId: string,
5
+
did: string,
6
+
sourcePlatform: string,
7
+
totalUsers: number,
8
+
matchedUsers: number
9
+
) {
10
+
const sql = getDbClient();
11
+
await sql`
12
+
INSERT INTO user_uploads (upload_id, did, source_platform, total_users, matched_users, unmatched_users)
13
+
VALUES (${uploadId}, ${did}, ${sourcePlatform}, ${totalUsers}, ${matchedUsers}, ${totalUsers - matchedUsers})
14
+
ON CONFLICT (upload_id) DO NOTHING
15
+
`;
6
16
}
7
17
8
-
// Get or create a source account, returns the source_account_id
9
18
export async function getOrCreateSourceAccount(
10
-
platform: string,
11
-
username: string
19
+
sourcePlatform: string,
20
+
sourceUsername: string
12
21
): Promise<number> {
13
22
const sql = getDbClient();
14
-
const normalized = normalizeUsername(username);
23
+
const normalized = sourceUsername.toLowerCase().replace(/[._-]/g, '');
15
24
16
25
const result = await sql`
17
26
INSERT INTO source_accounts (source_platform, source_username, normalized_username)
18
-
VALUES (${platform}, ${username}, ${normalized})
19
-
ON CONFLICT (source_platform, normalized_username)
20
-
DO UPDATE SET source_username = ${username}
27
+
VALUES (${sourcePlatform}, ${sourceUsername}, ${normalized})
28
+
ON CONFLICT (source_platform, normalized_username) DO UPDATE SET
29
+
source_username = ${sourceUsername}
21
30
RETURNING id
22
31
`;
23
32
24
-
return (result as Array<{ id: number }>)[0].id;
33
+
return (result as any[])[0].id;
25
34
}
26
35
27
-
// Link a user to a source account
28
36
export async function linkUserToSourceAccount(
29
37
uploadId: string,
30
38
did: string,
31
39
sourceAccountId: number,
32
-
sourceDate?: string
33
-
): Promise<void> {
40
+
sourceDate: string
41
+
) {
34
42
const sql = getDbClient();
35
-
36
43
await sql`
37
44
INSERT INTO user_source_follows (upload_id, did, source_account_id, source_date)
38
-
VALUES (${uploadId}, ${did}, ${sourceAccountId}, ${sourceDate || null})
45
+
VALUES (${uploadId}, ${did}, ${sourceAccountId}, ${sourceDate})
39
46
ON CONFLICT (upload_id, source_account_id) DO NOTHING
40
47
`;
41
48
}
42
49
43
-
// Store ATProto match for account (handles duplicates), returns atproto_match_id
44
50
export async function storeAtprotoMatch(
45
51
sourceAccountId: number,
46
52
atprotoDid: string,
47
53
atprotoHandle: string,
48
-
displayName: string | undefined,
49
-
avatar: string | undefined,
54
+
atprotoDisplayName: string | undefined,
55
+
atprotoAvatar: string | undefined,
50
56
matchScore: number
51
57
): Promise<number> {
52
58
const sql = getDbClient();
53
-
54
59
const result = await sql`
55
60
INSERT INTO atproto_matches (
56
-
source_account_id,
57
-
atproto_did,
58
-
atproto_handle,
59
-
atproto_display_name,
60
-
atproto_avatar,
61
-
match_score,
62
-
last_verified
61
+
source_account_id, atproto_did, atproto_handle,
62
+
atproto_display_name, atproto_avatar, match_score
63
63
)
64
64
VALUES (
65
-
${sourceAccountId},
66
-
${atprotoDid},
67
-
${atprotoHandle},
68
-
${displayName || null},
69
-
${avatar || null},
70
-
${matchScore},
71
-
NOW()
65
+
${sourceAccountId}, ${atprotoDid}, ${atprotoHandle},
66
+
${atprotoDisplayName || null}, ${atprotoAvatar || null}, ${matchScore}
72
67
)
73
-
ON CONFLICT (source_account_id, atproto_did)
74
-
DO UPDATE SET
68
+
ON CONFLICT (source_account_id, atproto_did) DO UPDATE SET
75
69
atproto_handle = ${atprotoHandle},
76
-
atproto_display_name = ${displayName || null},
77
-
atproto_avatar = ${avatar || null},
70
+
atproto_display_name = ${atprotoDisplayName || null},
71
+
atproto_avatar = ${atprotoAvatar || null},
78
72
match_score = ${matchScore},
79
73
last_verified = NOW()
80
74
RETURNING id
81
75
`;
82
76
83
-
return (result as Array<{ id: number }>)[0].id;
77
+
return (result as any[])[0].id;
84
78
}
85
79
86
-
// Mark source account as having matches
87
-
export async function markSourceAccountMatched(sourceAccountId: number): Promise<void> {
80
+
export async function markSourceAccountMatched(sourceAccountId: number) {
88
81
const sql = getDbClient();
89
-
90
82
await sql`
91
83
UPDATE source_accounts
92
-
SET match_found = TRUE, match_found_at = NOW()
84
+
SET match_found = true, match_found_at = NOW()
93
85
WHERE id = ${sourceAccountId}
94
86
`;
95
87
}
96
88
97
-
// Create user match status (tracks if user has viewed/followed this match)
98
89
export async function createUserMatchStatus(
99
90
did: string,
100
91
atprotoMatchId: number,
101
92
sourceAccountId: number,
102
-
viewed: boolean = true
103
-
): Promise<void> {
93
+
viewed: boolean = false
94
+
) {
104
95
const sql = getDbClient();
105
-
106
96
await sql`
107
97
INSERT INTO user_match_status (did, atproto_match_id, source_account_id, viewed, viewed_at)
108
-
VALUES (
109
-
${did},
110
-
${atprotoMatchId},
111
-
${sourceAccountId},
112
-
${viewed},
113
-
${viewed ? 'NOW()' : null}
114
-
)
115
-
ON CONFLICT (did, atproto_match_id) DO NOTHING
98
+
VALUES (${did}, ${atprotoMatchId}, ${sourceAccountId}, ${viewed}, ${viewed ? 'NOW()' : null})
99
+
ON CONFLICT (did, atproto_match_id) DO UPDATE SET
100
+
viewed = ${viewed},
101
+
viewed_at = CASE WHEN ${viewed} THEN NOW() ELSE user_match_status.viewed_at END
102
+
`;
103
+
}
104
+
105
+
// NEW: Bulk operations for Phase 2
106
+
export async function bulkCreateSourceAccounts(
107
+
sourcePlatform: string,
108
+
usernames: string[]
109
+
): Promise<Map<string, number>> {
110
+
const sql = getDbClient();
111
+
112
+
// Prepare bulk insert values
113
+
const values = usernames.map(username => ({
114
+
platform: sourcePlatform,
115
+
username: username,
116
+
normalized: username.toLowerCase().replace(/[._-]/g, '')
117
+
}));
118
+
119
+
// Build bulk insert query with unnest
120
+
const platforms = values.map(v => v.platform);
121
+
const source_usernames = values.map(v => v.username);
122
+
const normalized = values.map(v => v.normalized);
123
+
124
+
const result = await sql`
125
+
INSERT INTO source_accounts (source_platform, source_username, normalized_username)
126
+
SELECT *
127
+
FROM UNNEST(
128
+
${platforms}::text[],
129
+
${source_usernames}::text[],
130
+
${normalized}::text[]
131
+
) AS t(source_platform, source_username, normalized_username)
132
+
ON CONFLICT (source_platform, normalized_username) DO UPDATE
133
+
SET source_username = EXCLUDED.source_username
134
+
RETURNING id, normalized_username
116
135
`;
136
+
137
+
138
+
// Create map of normalized username to ID
139
+
const idMap = new Map<string, number>();
140
+
for (const row of result as any[]) {
141
+
idMap.set(row.normalized_username, row.id);
142
+
}
143
+
144
+
return idMap;
117
145
}
118
146
119
-
// Create upload record
120
-
export async function createUpload(
147
+
// ==================== THIS FUNCTION IS NOW FIXED ====================
148
+
export async function bulkLinkUserToSourceAccounts(
121
149
uploadId: string,
122
150
did: string,
123
-
platform: string,
124
-
totalUsers: number,
125
-
matchedUsers: number
126
-
): Promise<void> {
151
+
links: Array<{ sourceAccountId: number; sourceDate: string }>
152
+
) {
127
153
const sql = getDbClient();
128
154
155
+
const numLinks = links.length;
156
+
if (numLinks === 0) return;
157
+
158
+
// Extract arrays for columns that change
159
+
const sourceAccountIds = links.map(l => l.sourceAccountId);
160
+
const sourceDates = links.map(l => l.sourceDate);
161
+
162
+
// Create arrays for the static columns
163
+
const uploadIds = Array(numLinks).fill(uploadId);
164
+
const dids = Array(numLinks).fill(did);
165
+
166
+
// Use the parallel UNNEST pattern, which is proven to work in other functions
129
167
await sql`
130
-
INSERT INTO user_uploads (upload_id, did, source_platform, total_users, matched_users, unmatched_users)
131
-
VALUES (
132
-
${uploadId},
133
-
${did},
134
-
${platform},
135
-
${totalUsers},
136
-
${matchedUsers},
137
-
${totalUsers - matchedUsers}
168
+
INSERT INTO user_source_follows (upload_id, did, source_account_id, source_date)
169
+
SELECT * FROM UNNEST(
170
+
${uploadIds}::text[],
171
+
${dids}::text[],
172
+
${sourceAccountIds}::integer[],
173
+
${sourceDates}::text[]
174
+
) AS t(upload_id, did, source_account_id, source_date)
175
+
ON CONFLICT (upload_id, source_account_id) DO NOTHING
176
+
`;
177
+
}
178
+
// ====================================================================
179
+
180
+
export async function bulkStoreAtprotoMatches(
181
+
matches: Array<{
182
+
sourceAccountId: number;
183
+
atprotoDid: string;
184
+
atprotoHandle: string;
185
+
atprotoDisplayName?: string;
186
+
atprotoAvatar?: string;
187
+
matchScore: number;
188
+
}>
189
+
): Promise<Map<string, number>> {
190
+
const sql = getDbClient();
191
+
192
+
if (matches.length === 0) return new Map();
193
+
194
+
const sourceAccountId = matches.map(m => m.sourceAccountId)
195
+
const atprotoDid = matches.map(m => m.atprotoDid)
196
+
const atprotoHandle = matches.map(m => m.atprotoHandle)
197
+
const atprotoDisplayName = matches.map(m => m.atprotoDisplayName || null)
198
+
const atprotoAvatar = matches.map(m => m.atprotoAvatar || null)
199
+
const matchScore = matches.map(m => m.matchScore)
200
+
201
+
const result = await sql`
202
+
INSERT INTO atproto_matches (
203
+
source_account_id, atproto_did, atproto_handle,
204
+
atproto_display_name, atproto_avatar, match_score
205
+
)
206
+
SELECT * FROM UNNEST(
207
+
${sourceAccountId}::integer[],
208
+
${atprotoDid}::text[],
209
+
${atprotoHandle}::text[],
210
+
${atprotoDisplayName}::text[],
211
+
${atprotoAvatar}::text[],
212
+
${matchScore}::integer[]
213
+
) AS t(
214
+
source_account_id, atproto_did, atproto_handle,
215
+
atproto_display_name, atproto_avatar, match_score
138
216
)
217
+
ON CONFLICT (source_account_id, atproto_did) DO UPDATE SET
218
+
atproto_handle = EXCLUDED.atproto_handle,
219
+
atproto_display_name = EXCLUDED.atproto_display_name,
220
+
atproto_avatar = EXCLUDED.atproto_avatar,
221
+
match_score = EXCLUDED.match_score,
222
+
last_verified = NOW()
223
+
RETURNING id, source_account_id, atproto_did
139
224
`;
225
+
226
+
// Create map of "sourceAccountId:atprotoDid" to match ID
227
+
const idMap = new Map<string, number>();
228
+
for (const row of result as any[]) {
229
+
idMap.set(`${row.source_account_id}:${row.atproto_did}`, row.id);
230
+
}
231
+
232
+
return idMap;
140
233
}
141
234
142
-
// Get user's uploads
143
-
export async function getUserUploads(did: string, platform?: string) {
235
+
export async function bulkMarkSourceAccountsMatched(sourceAccountIds: number[]) {
144
236
const sql = getDbClient();
145
237
146
-
if (platform) {
147
-
return await sql`
148
-
SELECT * FROM user_uploads
149
-
WHERE did = ${did} AND source_platform = ${platform}
150
-
ORDER BY created_at DESC
151
-
`;
152
-
}
238
+
if (sourceAccountIds.length === 0) return;
153
239
154
-
return await sql`
155
-
SELECT * FROM user_uploads
156
-
WHERE did = ${did}
157
-
ORDER BY created_at DESC
240
+
await sql`
241
+
UPDATE source_accounts
242
+
SET match_found = true, match_found_at = NOW()
243
+
WHERE id = ANY(${sourceAccountIds})
244
+
`;
245
+
}
246
+
247
+
export async function bulkCreateUserMatchStatus(
248
+
statuses: Array<{
249
+
did: string;
250
+
atprotoMatchId: number;
251
+
sourceAccountId: number;
252
+
viewed: boolean;
253
+
}>
254
+
) {
255
+
const sql = getDbClient();
256
+
257
+
if (statuses.length === 0) return;
258
+
259
+
const did = statuses.map(s => s.did)
260
+
const atprotoMatchId = statuses.map(s => s.atprotoMatchId)
261
+
const sourceAccountId = statuses.map(s => s.sourceAccountId)
262
+
const viewedFlags = statuses.map(s => s.viewed);
263
+
const viewedDates = statuses.map(s => s.viewed ? new Date() : null);
264
+
265
+
await sql`
266
+
INSERT INTO user_match_status (did, atproto_match_id, source_account_id, viewed, viewed_at)
267
+
SELECT * FROM UNNEST(
268
+
${did}::text[],
269
+
${atprotoMatchId}::integer[],
270
+
${sourceAccountId}::integer[],
271
+
${viewedFlags}::boolean[],
272
+
${viewedDates}::timestamp[]
273
+
) AS t(did, atproto_match_id, source_account_id, viewed, viewed_at)
274
+
ON CONFLICT (did, atproto_match_id) DO UPDATE SET
275
+
viewed = EXCLUDED.viewed,
276
+
viewed_at = CASE WHEN EXCLUDED.viewed THEN NOW() ELSE user_match_status.viewed_at END
158
277
`;
159
278
}
+89
-63
netlify/functions/save-results.ts
+89
-63
netlify/functions/save-results.ts
···
2
2
import { userSessions } from './oauth-stores-db';
3
3
import cookie from 'cookie';
4
4
import {
5
-
getOrCreateSourceAccount,
6
-
linkUserToSourceAccount,
7
-
storeAtprotoMatch,
8
-
markSourceAccountMatched,
9
-
createUserMatchStatus,
10
-
createUpload
5
+
createUpload,
6
+
bulkCreateSourceAccounts,
7
+
bulkLinkUserToSourceAccounts,
8
+
bulkStoreAtprotoMatches,
9
+
bulkMarkSourceAccountsMatched,
10
+
bulkCreateUserMatchStatus
11
11
} from './db-helpers';
12
12
import { getDbClient } from './db';
13
13
···
100
100
};
101
101
}
102
102
103
-
// IMPORTANT: Create upload record FIRST before processing results
104
-
// This is required because user_source_follows has a foreign key to user_uploads
103
+
// Create upload record FIRST
105
104
await createUpload(
106
105
uploadId,
107
106
userSession.did,
108
107
sourcePlatform,
109
108
results.length,
110
-
0 // We'll update this after processing
109
+
0
111
110
);
112
111
113
-
const BATCH_SIZE = 100;
114
-
const batches = [];
115
-
for (let i = 0; i < results.length; i += BATCH_SIZE) {
116
-
batches.push(results.slice(i, i + BATCH_SIZE));
117
-
}
118
-
119
-
for (const batch of batches) {
120
-
// Process batch in parallel
121
-
await Promise.all(batch.map(async (result) => {
122
-
try {
123
-
// 1. Get or create source account (handles race conditions)
124
-
const sourceAccountId = await getOrCreateSourceAccount(
125
-
sourcePlatform,
126
-
result.tiktokUser.username
127
-
);
128
-
129
-
// 2. Link this user to the source account
130
-
await linkUserToSourceAccount(
131
-
uploadId,
132
-
userSession.did,
112
+
// BULK OPERATION 1: Create all source accounts at once
113
+
const allUsernames = results.map(r => r.tiktokUser.username);
114
+
const sourceAccountIdMap = await bulkCreateSourceAccounts(sourcePlatform, allUsernames);
115
+
116
+
// BULK OPERATION 2: Link all users to source accounts
117
+
const links = results.map(result => {
118
+
const normalized = result.tiktokUser.username.toLowerCase().replace(/[._-]/g, '');
119
+
const sourceAccountId = sourceAccountIdMap.get(normalized);
120
+
return {
121
+
sourceAccountId: sourceAccountId!,
122
+
sourceDate: result.tiktokUser.date
123
+
};
124
+
}).filter(link => link.sourceAccountId !== undefined);
125
+
126
+
await bulkLinkUserToSourceAccounts(uploadId, userSession.did, links);
127
+
128
+
// BULK OPERATION 3: Store all atproto matches at once
129
+
const allMatches: Array<{
130
+
sourceAccountId: number;
131
+
atprotoDid: string;
132
+
atprotoHandle: string;
133
+
atprotoDisplayName?: string;
134
+
atprotoAvatar?: string;
135
+
matchScore: number;
136
+
}> = [];
137
+
138
+
const matchedSourceAccountIds: number[] = [];
139
+
140
+
for (const result of results) {
141
+
const normalized = result.tiktokUser.username.toLowerCase().replace(/[._-]/g, '');
142
+
const sourceAccountId = sourceAccountIdMap.get(normalized);
143
+
144
+
if (sourceAccountId && result.atprotoMatches && result.atprotoMatches.length > 0) {
145
+
matchedCount++;
146
+
matchedSourceAccountIds.push(sourceAccountId);
147
+
148
+
for (const match of result.atprotoMatches) {
149
+
allMatches.push({
133
150
sourceAccountId,
134
-
result.tiktokUser.date
135
-
);
136
-
137
-
// 3. If matches found, store them
138
-
if (result.atprotoMatches && result.atprotoMatches.length > 0) {
139
-
matchedCount++;
140
-
141
-
// Mark source account as matched
142
-
await markSourceAccountMatched(sourceAccountId);
143
-
144
-
// Store each match
145
-
for (const match of result.atprotoMatches) {
146
-
const atprotoMatchId = await storeAtprotoMatch(
147
-
sourceAccountId,
148
-
match.did,
149
-
match.handle,
150
-
match.displayName,
151
-
match.avatar,
152
-
match.matchScore
153
-
);
154
-
155
-
// Create user match status (viewed = true since they just searched)
156
-
await createUserMatchStatus(
157
-
userSession.did,
158
-
atprotoMatchId,
159
-
sourceAccountId,
160
-
true
161
-
);
162
-
}
163
-
}
164
-
} catch (error) {
165
-
console.error(`Error processing result for ${result.tiktokUser.username}:`, error);
166
-
// Continue processing other results
151
+
atprotoDid: match.did,
152
+
atprotoHandle: match.handle,
153
+
atprotoDisplayName: match.displayName,
154
+
atprotoAvatar: match.avatar,
155
+
matchScore: match.matchScore
156
+
});
167
157
}
168
-
}));
158
+
}
159
+
}
160
+
161
+
// Store all matches in one operation
162
+
let matchIdMap = new Map<string, number>();
163
+
if (allMatches.length > 0) {
164
+
matchIdMap = await bulkStoreAtprotoMatches(allMatches);
165
+
}
166
+
167
+
// BULK OPERATION 4: Mark all matched source accounts
168
+
if (matchedSourceAccountIds.length > 0) {
169
+
await bulkMarkSourceAccountsMatched(matchedSourceAccountIds);
170
+
}
171
+
172
+
// BULK OPERATION 5: Create all user match statuses
173
+
const statuses: Array<{
174
+
did: string;
175
+
atprotoMatchId: number;
176
+
sourceAccountId: number;
177
+
viewed: boolean;
178
+
}> = [];
179
+
180
+
for (const match of allMatches) {
181
+
const key = `${match.sourceAccountId}:${match.atprotoDid}`;
182
+
const matchId = matchIdMap.get(key);
183
+
if (matchId) {
184
+
statuses.push({
185
+
did: userSession.did,
186
+
atprotoMatchId: matchId,
187
+
sourceAccountId: match.sourceAccountId,
188
+
viewed: true
189
+
});
190
+
}
191
+
}
192
+
193
+
if (statuses.length > 0) {
194
+
await bulkCreateUserMatchStatus(statuses);
169
195
}
170
196
171
197
// Update upload record with final counts