11import config from '../config' ;
2+ import { redisCount } from '../util/utility' ;
23import { Archive } from './archive' ;
3- import {
4- getFullPlayerMatchesWithMetadata ,
5- getMatchDataFromBlobWithMetadata ,
6- } from './queries' ;
74import db from './db' ;
85import redis from './redis' ;
9- import cassandra from './cassandra' ;
10- import type { PutObjectCommandOutput } from '@aws-sdk/client-s3' ;
11- import { isDataComplete , redisCount } from '../util/utility' ;
12- import QueryStream from 'pg-query-stream' ;
13- import { Client } from 'pg' ;
14- import crypto from 'crypto' ;
156
16- const matchArchive = new Archive ( 'match' ) ;
17- const playerArchive = new Archive ( 'player' ) ;
7+ const matchArchive = config . ENABLE_MATCH_ARCHIVE ? new Archive ( 'match' ) : null ;
8+ const playerArchive = config . ENABLE_PLAYER_ARCHIVE ? new Archive ( 'player' ) : null ;
189
19- export async function doArchivePlayerMatches (
20- accountId : number ,
21- ) : Promise < PutObjectCommandOutput | { message : string } | null > {
22- if ( ! config . ENABLE_PLAYER_ARCHIVE ) {
23- return null ;
24- }
25- // Fetch our combined list of archive and current, selecting all fields
26- const full = await getFullPlayerMatchesWithMetadata ( accountId ) ;
27- const toArchive = full [ 0 ] ;
28- console . log ( full [ 1 ] ) ;
29- toArchive . forEach ( ( m , i ) => {
30- Object . keys ( m ) . forEach ( ( key ) => {
31- if ( m [ key as keyof ParsedPlayerMatch ] === null ) {
32- // Remove any null values from the matches for storage
33- delete m [ key as keyof ParsedPlayerMatch ] ;
34- }
35- } ) ;
36- } ) ;
37- // TODO (howard) Make sure the new list is longer than the old list
38- // Make sure we're archiving at least 1 match
39- if ( ! toArchive . length ) {
40- return null ;
41- }
42- // Put the blob
43- return playerArchive . archivePut (
44- accountId . toString ( ) ,
45- Buffer . from ( JSON . stringify ( toArchive ) ) ,
46- ) ;
47- // TODO (howard) delete the archived values from player_caches
48- // TODO (howard) keep the 20 highest match IDs for recentMatches
49- // TODO (howard) mark the user archived so we don't need to query archive on every request
50- // TODO (howard) add redis counts
51- }
52-
53- async function doArchiveFromBlob ( matchId : number ) {
54- if ( ! config . ENABLE_MATCH_ARCHIVE ) {
55- return ;
56- }
57- // Don't backfill when determining whether to archive
58- const [ match , metadata ] = await getMatchDataFromBlobWithMetadata (
59- matchId ,
60- false ,
61- ) ;
62- if ( ! match ) {
63- // Invalid/not found, skip
64- return ;
65- }
66- if ( metadata ?. has_api && ! metadata ?. has_gcdata && ! metadata ?. has_parsed ) {
67- // if it only contains API data, delete?
68- // If the match is old we might not be able to get back ability builds, HD/TD/HH
69- // We might also drop gcdata, identity, and ranks here
70- // await deleteMatch(matchId);
71- // console.log('DELETE match %s, apionly', matchId);
72- return ;
73- }
74- if ( metadata ?. has_parsed ) {
75- const isArchived = Boolean (
76- (
77- await db . raw (
78- 'select match_id from parsed_matches where match_id = ? and is_archived IS TRUE' ,
79- [ matchId ] ,
80- )
81- ) . rows [ 0 ] ,
82- ) ;
83- if ( isArchived ) {
84- console . log ( 'ALREADY ARCHIVED match %s' , matchId ) ;
85- await deleteParsed ( matchId ) ;
86- return ;
87- }
88- // check data completeness with isDataComplete
89- if ( ! isDataComplete ( match as ParsedMatch ) ) {
90- redisCount ( redis , 'incomplete_archive' ) ;
91- console . log ( 'INCOMPLETE match %s' , matchId ) ;
92- return ;
93- }
94- redisCount ( redis , 'match_archive_write' ) ;
95- // console.log('SIMULATE ARCHIVE match %s', matchId);
96- // TODO (howard) don't actually archive until verification of data format
97- return ;
98- // Archive the data since it's parsed. This might also contain api and gcdata
99- const blob = Buffer . from ( JSON . stringify ( match ) ) ;
100- const result = await matchArchive . archivePut ( matchId . toString ( ) , blob ) ;
101- if ( result ) {
102- // Mark the match archived
103- await db . raw (
104- `UPDATE parsed_matches SET is_archived = TRUE WHERE match_id = ?` ,
105- [ matchId ] ,
106- ) ;
107- // Delete the parsed data (this keeps the api and gcdata around in Cassandra since it doesn't take a ton of space)
108- await deleteParsed ( matchId ) ;
109- console . log ( 'ARCHIVE match %s, parsed' , matchId ) ;
110- }
111- return result ;
112- }
113- // if it's something else, e.g. contains api and gcdata only, leave it for now
114- // console.log('SKIP match %s, unparsed', matchId);
115- return ;
116- }
117-
118- async function deleteParsed ( matchId : number ) {
119- await cassandra . execute (
120- 'DELETE parsed from match_blobs WHERE match_id = ?' ,
121- [ matchId ] ,
122- {
123- prepare : true ,
124- } ,
125- ) ;
126- }
127-
128- export async function archivePostgresStream ( ) {
129- // Archive parsed matches that aren't archived from postgres records
130- const max = await getCurrentMaxArchiveID ( ) ;
131- const query = new QueryStream (
132- `
133- SELECT match_id
134- from parsed_matches
135- WHERE is_archived IS NULL
136- and match_id < ?
137- ORDER BY match_id asc` ,
138- [ max ] ,
139- ) ;
140- const pg = new Client ( config . POSTGRES_URL ) ;
141- await pg . connect ( ) ;
142- const stream = pg . query ( query ) ;
143- let i = 0 ;
144- stream . on ( 'readable' , async ( ) => {
145- let row ;
146- while ( ( row = stream . read ( ) ) ) {
147- i += 1 ;
148- console . log ( i ) ;
149- try {
150- await doArchiveFromBlob ( row . match_id ) ;
151- } catch ( e ) {
152- console . error ( e ) ;
153- }
154- }
155- } ) ;
156- stream . on ( 'end' , async ( ) => {
157- await pg . end ( ) ;
158- } ) ;
159- }
160-
161- async function archiveSequential ( start : number , max : number ) {
162- // Archive sequentially starting at a given ID (not all IDs may be valid)
163- for ( let i = start ; i < max ; i ++ ) {
164- console . log ( i ) ;
165- try {
166- await doArchiveFromBlob ( i ) ;
167- } catch ( e ) {
168- console . error ( e ) ;
169- }
170- }
171- }
172-
173- async function archiveRandom ( max : number ) {
174- const rand = randomInt ( 0 , max ) ;
175- // Bruteforce 1000 IDs starting at a random value (not all IDs may be valid)
176- const page = [ ] ;
177- for ( let i = 0 ; i < 1000 ; i ++ ) {
178- page . push ( rand + i ) ;
179- }
180- console . log ( page [ 0 ] ) ;
181- await Promise . allSettled ( page . map ( ( i ) => doArchiveFromBlob ( i ) ) ) ;
182- }
183-
184- export async function archiveToken ( max : number ) {
185- // Archive random matches from Cassandra using token range (not all may be parsed)
186- let page = await getTokenRange ( 1000 ) ;
187- page = page . filter ( ( id ) => id < max ) ;
188- console . log ( page [ 0 ] ) ;
189- await Promise . allSettled ( page . map ( ( i ) => doArchiveFromBlob ( i ) ) ) ;
190- }
191-
192- function randomBigInt ( byteCount : number ) {
193- return BigInt ( `0x${ crypto . randomBytes ( byteCount ) . toString ( 'hex' ) } ) ;
194- }
195-
196- function randomInt ( min : number , max : number ) {
197- return Math . floor ( Math . random ( ) * ( max - min ) + min ) ;
198- }
199-
200- export async function getCurrentMaxArchiveID ( ) {
201- // Get the current max_match_id from postgres, subtract 200000000
202- const max = ( await db . raw ( 'select max(match_id) from public_matches' ) )
203- ?. rows ?. [ 0 ] ?. max ;
204- const limit = max - 100000000 ;
205- return limit ;
206- }
207-
208- async function getTokenRange ( size : number ) {
209- // Convert to signed 64-bit integer
210- const signedBigInt = BigInt . asIntN ( 64 , randomBigInt ( 8 ) ) ;
211- // Get a page of matches (efffectively random, but guaranteed sequential read on one node)
212- const result = await cassandra . execute (
213- 'select match_id, token(match_id) from match_blobs where token(match_id) >= ? limit ? ALLOW FILTERING;' ,
214- [ signedBigInt . toString ( ) , size ] ,
215- {
216- prepare : true ,
217- fetchSize : size ,
218- autoPage : true ,
219- } ,
220- ) ;
221- return result . rows . map ( ( row ) => Number ( row . match_id ) ) ;
222- }
223-
224- export async function readArchivedPlayerMatches (
10+ export async function tryReadArchivedPlayerMatches (
22511 accountId : number ,
22612) : Promise < ParsedPlayerMatch [ ] > {
13+ if ( ! playerArchive ) {
14+ return [ ] ;
15+ }
22716 console . time ( 'archive:' + accountId ) ;
22817 const blob = await playerArchive . archiveGet ( accountId . toString ( ) ) ;
22918 const arr = blob ? JSON . parse ( blob . toString ( ) ) : [ ] ;
@@ -240,7 +29,7 @@ export async function tryReadArchivedMatch(
24029 matchId : number ,
24130) : Promise < ParsedMatch | null > {
24231 try {
243- if ( ! config . ENABLE_MATCH_ARCHIVE ) {
32+ if ( ! matchArchive ) {
24433 return null ;
24534 }
24635 // Check if the parsed data is archived
0 commit comments