+16 -16

apps/api/drizzle/0000_polite_pixie.sql apps/api/drizzle/0000_left_swordsman.sql

··· 4 4 "track_id" text NOT NULL, 5 5 "xata_createdat" timestamp DEFAULT now() NOT NULL, 6 6 "xata_updatedat" timestamp DEFAULT now() NOT NULL, 7 7 - "xata_version" integer NOT NULL 7 7 + "xata_version" integer 8 8 ); 9 9 --> statement-breakpoint 10 10 CREATE TABLE "albums" ( ··· 50 50 "album_id" text NOT NULL, 51 51 "xata_createdat" timestamp DEFAULT now() NOT NULL, 52 52 "xata_updatedat" timestamp DEFAULT now() NOT NULL, 53 53 - "xata_version" integer NOT NULL 53 53 + "xata_version" integer 54 54 ); 55 55 --> statement-breakpoint 56 56 CREATE TABLE "artist_tracks" ( ··· 59 59 "track_id" text NOT NULL, 60 60 "xata_createdat" timestamp DEFAULT now() NOT NULL, 61 61 "xata_updatedat" timestamp DEFAULT now() NOT NULL, 62 62 - "xata_version" integer NOT NULL 62 62 + "xata_version" integer 63 63 ); 64 64 --> statement-breakpoint 65 65 CREATE TABLE "artists" ( ··· 89 89 "email" text NOT NULL, 90 90 "is_beta_user" boolean DEFAULT false NOT NULL, 91 91 "user_id" text NOT NULL, 92 92 - "xata_version" text NOT NULL, 92 92 + "xata_version" text, 93 93 "xata_createdat" timestamp DEFAULT now() NOT NULL, 94 94 "xata_updatedat" timestamp DEFAULT now() NOT NULL, 95 95 CONSTRAINT "dropbox_accounts_email_unique" UNIQUE("email") ··· 102 102 "parent_id" text, 103 103 "dropbox_id" text NOT NULL, 104 104 "file_id" text NOT NULL, 105 105 - "xata_version" text NOT NULL, 105 105 + "xata_version" text, 106 106 "xata_createdat" timestamp DEFAULT now() NOT NULL, 107 107 "xata_updatedat" timestamp DEFAULT now() NOT NULL, 108 108 CONSTRAINT "dropbox_directories_file_id_unique" UNIQUE("file_id") ··· 116 116 "track_id" text NOT NULL, 117 117 "directory_id" text, 118 118 "file_id" text NOT NULL, 119 119 - "xata_version" text NOT NULL, 119 119 + "xata_version" text, 120 120 "xata_createdat" timestamp DEFAULT now() NOT NULL, 121 121 "xata_updatedat" timestamp DEFAULT now() NOT NULL, 122 122 CONSTRAINT "dropbox_paths_file_id_unique" UNIQUE("file_id") ··· 133 133 "xata_id" text PRIMARY KEY DEFAULT xata_id(), 134 134 "user_id" text NOT NULL, 135 135 "dropbox_token_id" text NOT NULL, 136 136 - "xata_version" text NOT NULL, 136 136 + "xata_version" text, 137 137 "xata_createdat" timestamp DEFAULT now() NOT NULL, 138 138 "xata_updatedat" timestamp DEFAULT now() NOT NULL 139 139 ); ··· 143 143 "email" text NOT NULL, 144 144 "is_beta_user" boolean DEFAULT false NOT NULL, 145 145 "user_id" text NOT NULL, 146 146 - "xata_version" text NOT NULL, 146 146 + "xata_version" text, 147 147 "xata_createdat" timestamp DEFAULT now() NOT NULL, 148 148 "xata_updatedat" timestamp DEFAULT now() NOT NULL, 149 149 CONSTRAINT "google_drive_accounts_email_unique" UNIQUE("email") ··· 156 156 "parent_id" text, 157 157 "google_drive_id" text NOT NULL, 158 158 "file_id" text NOT NULL, 159 159 - "xata_version" text NOT NULL, 159 159 + "xata_version" text, 160 160 "xata_createdat" timestamp DEFAULT now() NOT NULL, 161 161 "xata_updatedat" timestamp DEFAULT now() NOT NULL, 162 162 CONSTRAINT "google_drive_directories_file_id_unique" UNIQUE("file_id") ··· 169 169 "name" text NOT NULL, 170 170 "directory_id" text, 171 171 "file_id" text NOT NULL, 172 172 - "xata_version" text NOT NULL, 172 172 + "xata_version" text, 173 173 "xata_createdat" timestamp DEFAULT now() NOT NULL, 174 174 "xata_updatedat" timestamp DEFAULT now() NOT NULL, 175 175 CONSTRAINT "google_drive_paths_file_id_unique" UNIQUE("file_id") ··· 186 186 "xata_id" text PRIMARY KEY DEFAULT xata_id(), 187 187 "google_drive_token_id" text NOT NULL, 188 188 "user_id" text NOT NULL, 189 189 - "xata_version" text NOT NULL, 189 189 + "xata_version" text, 190 190 "xata_createdat" timestamp DEFAULT now() NOT NULL, 191 191 "xata_updatedat" timestamp DEFAULT now() NOT NULL 192 192 ); ··· 287 287 --> statement-breakpoint 288 288 CREATE TABLE "spotify_accounts" ( 289 289 "xata_id" text PRIMARY KEY DEFAULT xata_id(), 290 290 - "xata_version" integer NOT NULL, 290 290 + "xata_version" integer, 291 291 "email" text NOT NULL, 292 292 "user_id" text NOT NULL, 293 293 "is_beta_user" boolean DEFAULT false NOT NULL, ··· 297 297 --> statement-breakpoint 298 298 CREATE TABLE "spotify_tokens" ( 299 299 "xata_id" text PRIMARY KEY DEFAULT xata_id(), 300 300 - "xata_version" integer NOT NULL, 300 300 + "xata_version" integer, 301 301 "access_token" text NOT NULL, 302 302 "refresh_token" text NOT NULL, 303 303 "user_id" text NOT NULL, ··· 347 347 "album_id" text NOT NULL, 348 348 "xata_createdat" timestamp DEFAULT now() NOT NULL, 349 349 "xata_updatedat" timestamp DEFAULT now() NOT NULL, 350 350 - "xata_version" integer NOT NULL, 350 350 + "xata_version" integer, 351 351 "scrobbles" integer, 352 352 "uri" text NOT NULL, 353 353 CONSTRAINT "user_albums_uri_unique" UNIQUE("uri") ··· 359 359 "artist_id" text NOT NULL, 360 360 "xata_createdat" timestamp DEFAULT now() NOT NULL, 361 361 "xata_updatedat" timestamp DEFAULT now() NOT NULL, 362 362 - "xata_version" integer NOT NULL, 362 362 + "xata_version" integer, 363 363 "scrobbles" integer, 364 364 "uri" text NOT NULL, 365 365 CONSTRAINT "user_artists_uri_unique" UNIQUE("uri") ··· 380 380 "track_id" text NOT NULL, 381 381 "xata_createdat" timestamp DEFAULT now() NOT NULL, 382 382 "xata_updatedat" timestamp DEFAULT now() NOT NULL, 383 383 - "xata_version" integer NOT NULL, 383 383 + "xata_version" integer, 384 384 "uri" text NOT NULL, 385 385 "scrobbles" integer, 386 386 CONSTRAINT "user_tracks_uri_unique" UNIQUE("uri")

+17 -17

apps/api/drizzle/meta/0000_snapshot.json

··· 1 1 { 2 2 - "id": "f7f59d87-3fa3-4146-91b5-692a483d2abc", 2 2 + "id": "6b33dcd0-52df-4403-bdc0-31517769923d", 3 3 "prevId": "00000000-0000-0000-0000-000000000000", 4 4 "version": "7", 5 5 "dialect": "postgresql", ··· 44 44 "name": "xata_version", 45 45 "type": "integer", 46 46 "primaryKey": false, 47 47 - "notNull": true 47 47 + "notNull": false 48 48 } 49 49 }, 50 50 "indexes": {}, ··· 360 360 "name": "xata_version", 361 361 "type": "integer", 362 362 "primaryKey": false, 363 363 - "notNull": true 363 363 + "notNull": false 364 364 } 365 365 }, 366 366 "indexes": {}, ··· 438 438 "name": "xata_version", 439 439 "type": "integer", 440 440 "primaryKey": false, 441 441 - "notNull": true 441 441 + "notNull": false 442 442 } 443 443 }, 444 444 "indexes": {}, ··· 641 641 "name": "xata_version", 642 642 "type": "text", 643 643 "primaryKey": false, 644 644 - "notNull": true 644 644 + "notNull": false 645 645 }, 646 646 "xata_createdat": { 647 647 "name": "xata_createdat", ··· 732 732 "name": "xata_version", 733 733 "type": "text", 734 734 "primaryKey": false, 735 735 - "notNull": true 735 735 + "notNull": false 736 736 }, 737 737 "xata_createdat": { 738 738 "name": "xata_createdat", ··· 829 829 "name": "xata_version", 830 830 "type": "text", 831 831 "primaryKey": false, 832 832 - "notNull": true 832 832 + "notNull": false 833 833 }, 834 834 "xata_createdat": { 835 835 "name": "xata_createdat", ··· 941 941 "name": "xata_version", 942 942 "type": "text", 943 943 "primaryKey": false, 944 944 - "notNull": true 944 944 + "notNull": false 945 945 }, 946 946 "xata_createdat": { 947 947 "name": "xata_createdat", ··· 1026 1026 "name": "xata_version", 1027 1027 "type": "text", 1028 1028 "primaryKey": false, 1029 1029 - "notNull": true 1029 1029 + "notNull": false 1030 1030 }, 1031 1031 "xata_createdat": { 1032 1032 "name": "xata_createdat", ··· 1117 1117 "name": "xata_version", 1118 1118 "type": "text", 1119 1119 "primaryKey": false, 1120 1120 - "notNull": true 1120 1120 + "notNull": false 1121 1121 }, 1122 1122 "xata_createdat": { 1123 1123 "name": "xata_createdat", ··· 1208 1208 "name": "xata_version", 1209 1209 "type": "text", 1210 1210 "primaryKey": false, 1211 1211 - "notNull": true 1211 1211 + "notNull": false 1212 1212 }, 1213 1213 "xata_createdat": { 1214 1214 "name": "xata_createdat", ··· 1320 1320 "name": "xata_version", 1321 1321 "type": "text", 1322 1322 "primaryKey": false, 1323 1323 - "notNull": true 1323 1323 + "notNull": false 1324 1324 }, 1325 1325 "xata_createdat": { 1326 1326 "name": "xata_createdat", ··· 2238 2238 "name": "xata_version", 2239 2239 "type": "integer", 2240 2240 "primaryKey": false, 2241 2241 - "notNull": true 2241 2241 + "notNull": false 2242 2242 }, 2243 2243 "email": { 2244 2244 "name": "email", ··· 2310 2310 "name": "xata_version", 2311 2311 "type": "integer", 2312 2312 "primaryKey": false, 2313 2313 - "notNull": true 2313 2313 + "notNull": false 2314 2314 }, 2315 2315 "access_token": { 2316 2316 "name": "access_token", ··· 2628 2628 "name": "xata_version", 2629 2629 "type": "integer", 2630 2630 "primaryKey": false, 2631 2631 - "notNull": true 2631 2631 + "notNull": false 2632 2632 }, 2633 2633 "scrobbles": { 2634 2634 "name": "scrobbles", ··· 2726 2726 "name": "xata_version", 2727 2727 "type": "integer", 2728 2728 "primaryKey": false, 2729 2729 - "notNull": true 2729 2729 + "notNull": false 2730 2730 }, 2731 2731 "scrobbles": { 2732 2732 "name": "scrobbles", ··· 2903 2903 "name": "xata_version", 2904 2904 "type": "integer", 2905 2905 "primaryKey": false, 2906 2906 - "notNull": true 2906 2906 + "notNull": false 2907 2907 }, 2908 2908 "uri": { 2909 2909 "name": "uri",

+2 -2

apps/api/drizzle/meta/_journal.json

··· 5 5 { 6 6 "idx": 0, 7 7 "version": "7", 8 8 - "when": 1759738835464, 9 9 - "tag": "0000_polite_pixie", 8 8 + "when": 1759744201158, 9 9 + "tag": "0000_left_swordsman", 10 10 "breakpoints": true 11 11 } 12 12 ]

+1 -1

apps/api/src/schema/album-tracks.ts

··· 13 13 .references(() => tracks.id), 14 14 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 15 15 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 16 16 - xataVersion: integer("xata_version").notNull(), 16 16 + xataVersion: integer("xata_version"), 17 17 }); 18 18 19 19 export type SelectAlbumTrack = InferSelectModel<typeof albumTracks>;

+1 -1

apps/api/src/schema/artist-albums.ts

··· 13 13 .references(() => albums.id), 14 14 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 15 15 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 16 16 - xataVersion: integer("xata_version").notNull(), 16 16 + xataVersion: integer("xata_version"), 17 17 }); 18 18 19 19 export type SelectArtistAlbum = InferSelectModel<typeof artistAlbums>;

+1 -1

apps/api/src/schema/artist-tracks.ts

··· 13 13 .references(() => tracks.id), 14 14 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 15 15 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 16 16 - xataVersion: integer("xata_version").notNull(), 16 16 + xataVersion: integer("xata_version"), 17 17 }); 18 18 19 19 export type SelectArtistTrack = InferSelectModel<typeof artistTracks>;

+1 -1

apps/api/src/schema/dropbox-accounts.ts

··· 9 9 userId: text("user_id") 10 10 .notNull() 11 11 .references(() => users.id), 12 12 - xataVersion: text("xata_version").notNull(), 12 12 + xataVersion: text("xata_version"), 13 13 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 14 14 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 15 15 });

+1 -1

apps/api/src/schema/dropbox-directories.ts

··· 8 8 parentId: text("parent_id").references(() => dropboxDirectories.id), 9 9 dropboxId: text("dropbox_id").notNull(), 10 10 fileId: text("file_id").notNull().unique(), 11 11 - xataVersion: text("xata_version").notNull(), 11 11 + xataVersion: text("xata_version"), 12 12 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 13 13 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 14 14 });

+1 -1

apps/api/src/schema/dropbox-paths.ts

··· 10 10 trackId: text("track_id").notNull(), 11 11 directoryId: text("directory_id").references(() => dropboxDirectories.id), 12 12 fileId: text("file_id").notNull().unique(), 13 13 - xataVersion: text("xata_version").notNull(), 13 13 + xataVersion: text("xata_version"), 14 14 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 15 15 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 16 16 });

+1 -1

apps/api/src/schema/dropbox.ts

··· 11 11 dropboxTokenId: text("dropbox_token_id") 12 12 .notNull() 13 13 .references(() => dropboxTokens.id), 14 14 - xataVersion: text("xata_version").notNull(), 14 14 + xataVersion: text("xata_version"), 15 15 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 16 16 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 17 17 });

+1 -1

apps/api/src/schema/google-drive-accounts.ts

··· 9 9 userId: text("user_id") 10 10 .notNull() 11 11 .references(() => users.id), 12 12 - xataVersion: text("xata_version").notNull(), 12 12 + xataVersion: text("xata_version"), 13 13 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 14 14 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 15 15 });

+1 -1

apps/api/src/schema/google-drive-directories.ts

··· 8 8 parentId: text("parent_id").references(() => googleDriveDirectories.id), 9 9 googleDriveId: text("google_drive_id").notNull(), 10 10 fileId: text("file_id").notNull().unique(), 11 11 - xataVersion: text("xata_version").notNull(), 11 11 + xataVersion: text("xata_version"), 12 12 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 13 13 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 14 14 });

+1 -1

apps/api/src/schema/google-drive-paths.ts

··· 9 9 name: text("name").notNull(), 10 10 directoryId: text("directory_id").references(() => googleDriveDirectories.id), 11 11 fileId: text("file_id").notNull().unique(), 12 12 - xataVersion: text("xata_version").notNull(), 12 12 + xataVersion: text("xata_version"), 13 13 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 14 14 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 15 15 });

+1 -1

apps/api/src/schema/googledrive.ts

··· 11 11 userId: text("user_id") 12 12 .notNull() 13 13 .references(() => users.id), 14 14 - xataVersion: text("xata_version").notNull(), 14 14 + xataVersion: text("xata_version"), 15 15 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 16 16 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 17 17 });

+1 -1

apps/api/src/schema/spotify-accounts.ts

··· 10 10 11 11 const spotifyAccounts = pgTable("spotify_accounts", { 12 12 id: text("xata_id").primaryKey(), 13 13 - xataVersion: integer("xata_version").notNull(), 13 13 + xataVersion: integer("xata_version"), 14 14 email: text("email").notNull(), 15 15 userId: text("user_id") 16 16 .notNull()

+1 -1

apps/api/src/schema/spotify-tokens.ts

··· 4 4 5 5 const spotifyTokens = pgTable("spotify_tokens", { 6 6 id: text("xata_id").primaryKey(), 7 7 - xataVersion: integer("xata_version").notNull(), 7 7 + xataVersion: integer("xata_version"), 8 8 accessToken: text("access_token").notNull(), 9 9 refreshToken: text("refresh_token").notNull(), 10 10 userId: text("user_id")

+1 -1

apps/api/src/schema/user-albums.ts

··· 13 13 .references(() => albums.id), 14 14 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 15 15 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 16 16 - xataVersion: integer("xata_version").notNull(), 16 16 + xataVersion: integer("xata_version"), 17 17 scrobbles: integer("scrobbles"), 18 18 uri: text("uri").unique().notNull(), 19 19 });

+1 -1

apps/api/src/schema/user-artists.ts

··· 13 13 .references(() => artists.id), 14 14 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 15 15 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 16 16 - xataVersion: integer("xata_version").notNull(), 16 16 + xataVersion: integer("xata_version"), 17 17 scrobbles: integer("scrobbles"), 18 18 uri: text("uri").unique().notNull(), 19 19 });

+1 -1

apps/api/src/schema/user-tracks.ts

··· 13 13 .references(() => tracks.id), 14 14 createdAt: timestamp("xata_createdat").defaultNow().notNull(), 15 15 updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), 16 16 - xataVersion: integer("xata_version").notNull(), 16 16 + xataVersion: integer("xata_version"), 17 17 uri: text("uri").unique().notNull(), 18 18 scrobbles: integer("scrobbles"), 19 19 });

+336

crates/pgpull/src/lib.rs

··· 297 297 playlist_sync.context("Playlist sync task failed")??; 298 298 scrobble_sync.context("Scrobble sync task failed")??; 299 299 300 300 + let pool_clone = pool.clone(); 301 301 + let dest_pool_clone = dest_pool.clone(); 302 302 + let album_track_sync = tokio::spawn(async move { 303 303 + let total_album_tracks: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM album_tracks") 304 304 + .fetch_one(&pool_clone) 305 305 + .await?; 306 306 + let total_album_tracks = total_album_tracks.0; 307 307 + tracing::info!(total = %total_album_tracks.magenta(), "Total album tracks to sync"); 308 308 + 309 309 + const BATCH_SIZE: usize = 1000; 310 310 + 311 311 + let start = 0; 312 312 + let mut i = 1; 313 313 + 314 314 + for offset in (start..total_album_tracks).step_by(BATCH_SIZE) { 315 315 + let album_tracks = 316 316 + repo::album::get_album_tracks(&pool_clone, offset as i64, BATCH_SIZE as i64) 317 317 + .await?; 318 318 + tracing::info!( 319 319 + offset = %offset.magenta(), 320 320 + end = %((offset + album_tracks.len() as i64).min(total_album_tracks)).magenta(), 321 321 + total = %total_album_tracks.magenta(), 322 322 + "Fetched album tracks" 323 323 + ); 324 324 + 325 325 + for album_track in &album_tracks { 326 326 + tracing::info!(album_id = %album_track.album_id.cyan(), track_id = %album_track.track_id.magenta(), i = %i.magenta(), total = %total_album_tracks.magenta(), "Inserting album track"); 327 327 + match repo::album::insert_album_track(&dest_pool_clone, album_track).await { 328 328 + Ok(_) => {} 329 329 + Err(e) => { 330 330 + tracing::error!(error = %e, "Failed to insert album track"); 331 331 + } 332 332 + } 333 333 + i += 1; 334 334 + } 335 335 + } 336 336 + Ok::<(), Error>(()) 337 337 + }); 338 338 + 339 339 + let pool_clone = pool.clone(); 340 340 + let dest_pool_clone = dest_pool.clone(); 341 341 + let artist_album_sync = tokio::spawn(async move { 342 342 + let total_artist_albums: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM artist_albums") 343 343 + .fetch_one(&pool_clone) 344 344 + .await?; 345 345 + let total_artist_albums = total_artist_albums.0; 346 346 + tracing::info!(total = %total_artist_albums.magenta(), "Total artist albums to sync"); 347 347 + 348 348 + const BATCH_SIZE: usize = 1000; 349 349 + 350 350 + let start = 0; 351 351 + let mut i = 1; 352 352 + 353 353 + for offset in (start..total_artist_albums).step_by(BATCH_SIZE) { 354 354 + let artist_albums = 355 355 + repo::artist::get_artist_albums(&pool_clone, offset as i64, BATCH_SIZE as i64) 356 356 + .await?; 357 357 + tracing::info!( 358 358 + offset = %offset.magenta(), 359 359 + end = %((offset + artist_albums.len() as i64).min(total_artist_albums)).magenta(), 360 360 + total = %total_artist_albums.magenta(), 361 361 + "Fetched artist albums" 362 362 + ); 363 363 + 364 364 + for artist_album in &artist_albums { 365 365 + tracing::info!(artist_id = %artist_album.artist_id.cyan(), album_id = %artist_album.album_id.magenta(), i = %i.magenta(), total = %total_artist_albums.magenta(), "Inserting artist album"); 366 366 + match repo::artist::insert_artist_album(&dest_pool_clone, artist_album).await { 367 367 + Ok(_) => {} 368 368 + Err(e) => { 369 369 + tracing::error!(error = %e, "Failed to insert artist album"); 370 370 + } 371 371 + } 372 372 + i += 1; 373 373 + } 374 374 + } 375 375 + Ok::<(), Error>(()) 376 376 + }); 377 377 + 378 378 + let pool_clone = pool.clone(); 379 379 + let dest_pool_clone = dest_pool.clone(); 380 380 + let artist_track_sync = tokio::spawn(async move { 381 381 + let total_artist_tracks: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM artist_tracks") 382 382 + .fetch_one(&pool_clone) 383 383 + .await?; 384 384 + let total_artist_tracks = total_artist_tracks.0; 385 385 + tracing::info!(total = %total_artist_tracks.magenta(), "Total artist tracks to sync"); 386 386 + const BATCH_SIZE: usize = 1000; 387 387 + 388 388 + let start = 0; 389 389 + let mut i = 1; 390 390 + 391 391 + for offset in (start..total_artist_tracks).step_by(BATCH_SIZE) { 392 392 + let artist_tracks = 393 393 + repo::artist::get_artist_tracks(&pool_clone, offset as i64, BATCH_SIZE as i64) 394 394 + .await?; 395 395 + tracing::info!( 396 396 + offset = %offset.magenta(), 397 397 + end = %((offset + artist_tracks.len() as i64).min(total_artist_tracks)).magenta(), 398 398 + total = %total_artist_tracks.magenta(), 399 399 + "Fetched artist tracks" 400 400 + ); 401 401 + 402 402 + for artist_track in &artist_tracks { 403 403 + tracing::info!(artist_id = %artist_track.artist_id.cyan(), track_id = %artist_track.track_id.magenta(), i = %i.magenta(), total = %total_artist_tracks.magenta(), "Inserting artist track"); 404 404 + match repo::artist::insert_artist_track(&dest_pool_clone, artist_track).await { 405 405 + Ok(_) => {} 406 406 + Err(e) => { 407 407 + tracing::error!(error = %e, "Failed to insert artist track"); 408 408 + } 409 409 + } 410 410 + i += 1; 411 411 + } 412 412 + } 413 413 + Ok::<(), Error>(()) 414 414 + }); 415 415 + 416 416 + let pool_clone = pool.clone(); 417 417 + let dest_pool_clone = dest_pool.clone(); 418 418 + let playlist_track_sync = tokio::spawn(async move { 419 419 + let total_playlist_tracks: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM playlist_tracks") 420 420 + .fetch_one(&pool_clone) 421 421 + .await?; 422 422 + let total_playlist_tracks = total_playlist_tracks.0; 423 423 + tracing::info!(total = %total_playlist_tracks.magenta(), "Total playlist tracks to sync"); 424 424 + 425 425 + const BATCH_SIZE: usize = 1000; 426 426 + 427 427 + let start = 0; 428 428 + let mut i = 1; 429 429 + 430 430 + for offset in (start..total_playlist_tracks).step_by(BATCH_SIZE) { 431 431 + let playlist_tracks = 432 432 + repo::playlist::get_playlist_tracks(&pool_clone, offset as i64, BATCH_SIZE as i64) 433 433 + .await?; 434 434 + tracing::info!( 435 435 + offset = %offset.magenta(), 436 436 + end = %((offset + playlist_tracks.len() as i64).min(total_playlist_tracks)).magenta(), 437 437 + total = %total_playlist_tracks.magenta(), 438 438 + "Fetched playlist tracks" 439 439 + ); 440 440 + 441 441 + for playlist_track in &playlist_tracks { 442 442 + tracing::info!(playlist_id = %playlist_track.playlist_id.cyan(), track_id = %playlist_track.track_id.magenta(), i = %i.magenta(), total = %total_playlist_tracks.magenta(), "Inserting playlist track"); 443 443 + match repo::playlist::insert_playlist_track(&dest_pool_clone, playlist_track).await 444 444 + { 445 445 + Ok(_) => {} 446 446 + Err(e) => { 447 447 + tracing::error!(error = %e, "Failed to insert playlist track"); 448 448 + } 449 449 + } 450 450 + i += 1; 451 451 + } 452 452 + } 453 453 + Ok::<(), Error>(()) 454 454 + }); 455 455 + 456 456 + let pool_clone = pool.clone(); 457 457 + let dest_pool_clone = dest_pool.clone(); 458 458 + let user_album_sync = tokio::spawn(async move { 459 459 + let total_user_albums: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM user_albums") 460 460 + .fetch_one(&pool_clone) 461 461 + .await?; 462 462 + let total_user_albums = total_user_albums.0; 463 463 + tracing::info!(total = %total_user_albums.magenta(), "Total user albums to sync"); 464 464 + const BATCH_SIZE: usize = 1000; 465 465 + 466 466 + let start = 0; 467 467 + let mut i = 1; 468 468 + 469 469 + for offset in (start..total_user_albums).step_by(BATCH_SIZE) { 470 470 + let user_albums = 471 471 + repo::album::get_user_albums(&pool_clone, offset as i64, BATCH_SIZE as i64).await?; 472 472 + tracing::info!( 473 473 + offset = %offset.magenta(), 474 474 + end = %((offset + user_albums.len() as i64).min(total_user_albums)).magenta(), 475 475 + total = %total_user_albums.magenta(), 476 476 + "Fetched user albums" 477 477 + ); 478 478 + 479 479 + for user_album in &user_albums { 480 480 + tracing::info!(user_id = %user_album.user_id.cyan(), album_id = %user_album.album_id.magenta(), i = %i.magenta(), total = %total_user_albums.magenta(), "Inserting user album"); 481 481 + match repo::album::insert_user_album(&dest_pool_clone, user_album).await { 482 482 + Ok(_) => {} 483 483 + Err(e) => { 484 484 + tracing::error!(error = %e, "Failed to insert user album"); 485 485 + } 486 486 + } 487 487 + i += 1; 488 488 + } 489 489 + } 490 490 + Ok::<(), Error>(()) 491 491 + }); 492 492 + 493 493 + let pool_clone = pool.clone(); 494 494 + let dest_pool_clone = dest_pool.clone(); 495 495 + let user_artist_sync = tokio::spawn(async move { 496 496 + let total_user_artists: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM user_artists") 497 497 + .fetch_one(&pool_clone) 498 498 + .await?; 499 499 + let total_user_artists = total_user_artists.0; 500 500 + tracing::info!(total = %total_user_artists.magenta(), "Total user artists to sync"); 501 501 + const BATCH_SIZE: usize = 1000; 502 502 + 503 503 + let start = 0; 504 504 + let mut i = 1; 505 505 + 506 506 + for offset in (start..total_user_artists).step_by(BATCH_SIZE) { 507 507 + let user_artists = 508 508 + repo::artist::get_user_artists(&pool_clone, offset as i64, BATCH_SIZE as i64) 509 509 + .await?; 510 510 + tracing::info!( 511 511 + offset = %offset.magenta(), 512 512 + end = %((offset + user_artists.len() as i64).min(total_user_artists)).magenta(), 513 513 + total = %total_user_artists.magenta(), 514 514 + "Fetched user artists" 515 515 + ); 516 516 + 517 517 + for user_artist in &user_artists { 518 518 + tracing::info!(user_id = %user_artist.user_id.cyan(), artist_id = %user_artist.artist_id.magenta(), i = %i.magenta(), total = %total_user_artists.magenta(), "Inserting user artist"); 519 519 + match repo::artist::insert_user_artist(&dest_pool_clone, user_artist).await { 520 520 + Ok(_) => {} 521 521 + Err(e) => { 522 522 + tracing::error!(error = %e, "Failed to insert user artist"); 523 523 + } 524 524 + } 525 525 + i += 1; 526 526 + } 527 527 + } 528 528 + Ok::<(), Error>(()) 529 529 + }); 530 530 + 531 531 + let pool_clone = pool.clone(); 532 532 + let dest_pool_clone = dest_pool.clone(); 533 533 + let user_track_sync = tokio::spawn(async move { 534 534 + let total_user_tracks: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM user_tracks") 535 535 + .fetch_one(&pool_clone) 536 536 + .await?; 537 537 + let total_user_tracks = total_user_tracks.0; 538 538 + tracing::info!(total = %total_user_tracks.magenta(), "Total user tracks to sync"); 539 539 + const BATCH_SIZE: usize = 1000; 540 540 + 541 541 + let start = 0; 542 542 + let mut i = 1; 543 543 + 544 544 + for offset in (start..total_user_tracks).step_by(BATCH_SIZE) { 545 545 + let user_tracks = 546 546 + repo::track::get_user_tracks(&pool_clone, offset as i64, BATCH_SIZE as i64).await?; 547 547 + tracing::info!( 548 548 + offset = %offset.magenta(), 549 549 + end = %((offset + user_tracks.len() as i64).min(total_user_tracks)).magenta(), 550 550 + total = %total_user_tracks.magenta(), 551 551 + "Fetched user tracks" 552 552 + ); 553 553 + 554 554 + for user_track in &user_tracks { 555 555 + tracing::info!(user_id = %user_track.user_id.cyan(), track_id = %user_track.track_id.magenta(), i = %i.magenta(), total = %total_user_tracks.magenta(), "Inserting user track"); 556 556 + match repo::track::insert_user_track(&dest_pool_clone, user_track).await { 557 557 + Ok(_) => {} 558 558 + Err(e) => { 559 559 + tracing::error!(error = %e, "Failed to insert user track"); 560 560 + } 561 561 + } 562 562 + i += 1; 563 563 + } 564 564 + } 565 565 + Ok::<(), Error>(()) 566 566 + }); 567 567 + 568 568 + let pool_clone = pool.clone(); 569 569 + let dest_pool_clone = dest_pool.clone(); 570 570 + let user_playlist_sync = tokio::spawn(async move { 571 571 + let total_user_playlists: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM user_playlists") 572 572 + .fetch_one(&pool_clone) 573 573 + .await?; 574 574 + 575 575 + let total_user_playlists = total_user_playlists.0; 576 576 + tracing::info!(total = %total_user_playlists.magenta(), "Total user playlists to sync"); 577 577 + const BATCH_SIZE: usize = 1000; 578 578 + 579 579 + let start = 0; 580 580 + let mut i = 1; 581 581 + 582 582 + for offset in (start..total_user_playlists).step_by(BATCH_SIZE) { 583 583 + let user_playlists = 584 584 + repo::playlist::get_user_playlists(&pool_clone, offset as i64, BATCH_SIZE as i64) 585 585 + .await?; 586 586 + tracing::info!( 587 587 + offset = %offset.magenta(), 588 588 + end = %((offset + user_playlists.len() as i64).min(total_user_playlists)).magenta(), 589 589 + total = %total_user_playlists.magenta(), 590 590 + "Fetched user playlists" 591 591 + ); 592 592 + 593 593 + for user_playlist in &user_playlists { 594 594 + tracing::info!(user_id = %user_playlist.user_id.cyan(), playlist_id = %user_playlist.playlist_id.magenta(), i = %i.magenta(), total = %total_user_playlists.magenta(), "Inserting user playlist"); 595 595 + match repo::playlist::insert_user_playlist(&dest_pool_clone, user_playlist).await { 596 596 + Ok(_) => {} 597 597 + Err(e) => { 598 598 + tracing::error!(error = %e, "Failed to insert user playlist"); 599 599 + } 600 600 + } 601 601 + i += 1; 602 602 + } 603 603 + } 604 604 + Ok::<(), Error>(()) 605 605 + }); 606 606 + 607 607 + let ( 608 608 + album_track_sync, 609 609 + artist_album_sync, 610 610 + artist_track_sync, 611 611 + playlist_track_sync, 612 612 + user_album_sync, 613 613 + user_artist_sync, 614 614 + user_track_sync, 615 615 + user_playlist_sync, 616 616 + ) = tokio::join!( 617 617 + album_track_sync, 618 618 + artist_album_sync, 619 619 + artist_track_sync, 620 620 + playlist_track_sync, 621 621 + user_album_sync, 622 622 + user_artist_sync, 623 623 + user_track_sync, 624 624 + user_playlist_sync 625 625 + ); 626 626 + 627 627 + album_track_sync.context("Album track sync task failed")??; 628 628 + artist_album_sync.context("Artist album sync task failed")??; 629 629 + artist_track_sync.context("Artist track sync task failed")??; 630 630 + playlist_track_sync.context("Playlist track sync task failed")??; 631 631 + user_album_sync.context("User album sync task failed")??; 632 632 + user_artist_sync.context("User artist sync task failed")??; 633 633 + user_track_sync.context("User track sync task failed")??; 634 634 + user_playlist_sync.context("User playlist sync task failed")??; 635 635 + 300 636 Ok(()) 301 637 }

+71 -1

crates/pgpull/src/repo/album.rs

··· 1 1 use anyhow::Error; 2 2 use sqlx::{Pool, Postgres}; 3 3 4 4 - use crate::xata::album::Album; 4 4 + use crate::xata::{album::Album, album_track::AlbumTrack, user_album::UserAlbum}; 5 5 6 6 pub async fn get_albums( 7 7 pool: &Pool<Postgres>, ··· 16 16 Ok(albums) 17 17 } 18 18 19 19 + pub async fn get_album_tracks( 20 20 + pool: &Pool<Postgres>, 21 21 + offset: i64, 22 22 + limit: i64, 23 23 + ) -> Result<Vec<AlbumTrack>, Error> { 24 24 + let album_tracks: Vec<AlbumTrack> = 25 25 + sqlx::query_as("SELECT * FROM album_tracks OFFSET $1 LIMIT $2") 26 26 + .bind(offset) 27 27 + .bind(limit) 28 28 + .fetch_all(pool) 29 29 + .await?; 30 30 + Ok(album_tracks) 31 31 + } 32 32 + 33 33 + pub async fn get_user_albums( 34 34 + pool: &Pool<Postgres>, 35 35 + offset: i64, 36 36 + limit: i64, 37 37 + ) -> Result<Vec<UserAlbum>, Error> { 38 38 + let user_albums: Vec<UserAlbum> = 39 39 + sqlx::query_as("SELECT * FROM user_albums OFFSET $1 LIMIT $2") 40 40 + .bind(offset) 41 41 + .bind(limit) 42 42 + .fetch_all(pool) 43 43 + .await?; 44 44 + Ok(user_albums) 45 45 + } 46 46 + 19 47 pub async fn insert_album(pool: &Pool<Postgres>, album: &Album) -> Result<(), Error> { 20 48 sqlx::query( 21 49 r#"INSERT INTO albums ( ··· 55 83 56 84 Ok(()) 57 85 } 86 86 + 87 87 + pub async fn insert_album_track( 88 88 + pool: &Pool<Postgres>, 89 89 + album_track: &AlbumTrack, 90 90 + ) -> Result<(), Error> { 91 91 + sqlx::query( 92 92 + r#"INSERT INTO album_tracks ( 93 93 + xata_id, 94 94 + album_id, 95 95 + track_id, 96 96 + xata_createdat 97 97 + ) VALUES ($1, $2, $3) 98 98 + ON CONFLICT (xata_id) DO NOTHING"#, 99 99 + ) 100 100 + .bind(&album_track.xata_id) 101 101 + .bind(&album_track.album_id) 102 102 + .bind(&album_track.track_id) 103 103 + .execute(pool) 104 104 + .await?; 105 105 + Ok(()) 106 106 + } 107 107 + 108 108 + pub async fn insert_user_album(pool: &Pool<Postgres>, user_album: &UserAlbum) -> Result<(), Error> { 109 109 + sqlx::query( 110 110 + r#"INSERT INTO user_albums ( 111 111 + xata_id, 112 112 + user_id, 113 113 + album_id, 114 114 + uri, 115 115 + xata_createdat 116 116 + ) VALUES ($1, $2, $3, $4, $5) 117 117 + ON CONFLICT (xata_id) DO NOTHING"#, 118 118 + ) 119 119 + .bind(&user_album.xata_id) 120 120 + .bind(&user_album.user_id) 121 121 + .bind(&user_album.album_id) 122 122 + .bind(&user_album.uri) 123 123 + .bind(user_album.xata_createdat) 124 124 + .execute(pool) 125 125 + .await?; 126 126 + Ok(()) 127 127 + }

+113 -1

crates/pgpull/src/repo/artist.rs

··· 1 1 use anyhow::Error; 2 2 use sqlx::{Pool, Postgres}; 3 3 4 4 - use crate::xata::artist::Artist; 4 4 + use crate::xata::{ 5 5 + artist::Artist, artist_album::ArtistAlbum, artist_track::ArtistTrack, user_artist::UserArtist, 6 6 + }; 5 7 6 8 pub async fn get_artists( 7 9 pool: &Pool<Postgres>, ··· 56 58 .await?; 57 59 Ok(()) 58 60 } 61 61 + 62 62 + pub async fn get_artist_albums( 63 63 + pool: &Pool<Postgres>, 64 64 + offset: i64, 65 65 + limit: i64, 66 66 + ) -> Result<Vec<ArtistAlbum>, Error> { 67 67 + let artist_albums = 68 68 + sqlx::query_as::<_, ArtistAlbum>("SELECT * FROM artist_albums OFFSET $1 LIMIT $2") 69 69 + .bind(offset) 70 70 + .bind(limit) 71 71 + .fetch_all(pool) 72 72 + .await?; 73 73 + Ok(artist_albums) 74 74 + } 75 75 + 76 76 + pub async fn get_artist_tracks( 77 77 + pool: &Pool<Postgres>, 78 78 + offset: i64, 79 79 + limit: i64, 80 80 + ) -> Result<Vec<ArtistTrack>, Error> { 81 81 + let artist_tracks = 82 82 + sqlx::query_as::<_, ArtistTrack>("SELECT * FROM artist_tracks OFFSET $1 LIMIT $2") 83 83 + .bind(offset) 84 84 + .bind(limit) 85 85 + .fetch_all(pool) 86 86 + .await?; 87 87 + Ok(artist_tracks) 88 88 + } 89 89 + 90 90 + pub async fn get_user_artists( 91 91 + pool: &Pool<Postgres>, 92 92 + offset: i64, 93 93 + limit: i64, 94 94 + ) -> Result<Vec<UserArtist>, Error> { 95 95 + let user_artists = 96 96 + sqlx::query_as::<_, UserArtist>("SELECT * FROM user_artists OFFSET $1 LIMIT $2") 97 97 + .bind(offset) 98 98 + .bind(limit) 99 99 + .fetch_all(pool) 100 100 + .await?; 101 101 + Ok(user_artists) 102 102 + } 103 103 + 104 104 + pub async fn insert_artist_album( 105 105 + pool: &Pool<Postgres>, 106 106 + artist_album: &ArtistAlbum, 107 107 + ) -> Result<(), Error> { 108 108 + sqlx::query( 109 109 + r#"INSERT INTO artist_albums ( 110 110 + xata_id, 111 111 + artist_id, 112 112 + album_id, 113 113 + xata_createdat 114 114 + ) VALUES ($1, $2, $3, $4) 115 115 + ON CONFLICT (xata_id) DO NOTHING"#, 116 116 + ) 117 117 + .bind(&artist_album.xata_id) 118 118 + .bind(&artist_album.artist_id) 119 119 + .bind(&artist_album.album_id) 120 120 + .bind(artist_album.xata_createdat) 121 121 + .execute(pool) 122 122 + .await?; 123 123 + Ok(()) 124 124 + } 125 125 + 126 126 + pub async fn insert_artist_track( 127 127 + pool: &Pool<Postgres>, 128 128 + artist_track: &ArtistTrack, 129 129 + ) -> Result<(), Error> { 130 130 + sqlx::query( 131 131 + r#"INSERT INTO artist_tracks ( 132 132 + xata_id, 133 133 + artist_id, 134 134 + track_id, 135 135 + xata_createdat 136 136 + ) VALUES ($1, $2, $3, $4) 137 137 + ON CONFLICT (xata_id) DO NOTHING"#, 138 138 + ) 139 139 + .bind(&artist_track.xata_id) 140 140 + .bind(&artist_track.artist_id) 141 141 + .bind(&artist_track.track_id) // Reusing album_id field for track_id 142 142 + .bind(artist_track.xata_createdat) 143 143 + .execute(pool) 144 144 + .await?; 145 145 + Ok(()) 146 146 + } 147 147 + 148 148 + pub async fn insert_user_artist( 149 149 + pool: &Pool<Postgres>, 150 150 + user_artist: &UserArtist, 151 151 + ) -> Result<(), Error> { 152 152 + sqlx::query( 153 153 + r#"INSERT INTO user_artists ( 154 154 + xata_id, 155 155 + user_id, 156 156 + artist_id, 157 157 + uri, 158 158 + xata_createdat 159 159 + ) VALUES ($1, $2, $3, $4, $5) 160 160 + ON CONFLICT (xata_id) DO NOTHING"#, 161 161 + ) 162 162 + .bind(&user_artist.xata_id) 163 163 + .bind(&user_artist.user_id) 164 164 + .bind(&user_artist.artist_id) 165 165 + .bind(&user_artist.uri) 166 166 + .bind(user_artist.xata_createdat) 167 167 + .execute(pool) 168 168 + .await?; 169 169 + Ok(()) 170 170 + }

+77 -1

crates/pgpull/src/repo/playlist.rs

··· 1 1 use anyhow::Error; 2 2 use sqlx::{Pool, Postgres}; 3 3 4 4 - use crate::xata::playlist::Playlist; 4 4 + use crate::xata::{playlist::Playlist, playlist_track::PlaylistTrack, user_playlist::UserPlaylist}; 5 5 6 6 pub async fn get_playlists( 7 7 pool: &Pool<Postgres>, ··· 16 16 Ok(playlists) 17 17 } 18 18 19 19 + pub async fn get_playlist_tracks( 20 20 + pool: &Pool<Postgres>, 21 21 + offset: i64, 22 22 + limit: i64, 23 23 + ) -> Result<Vec<PlaylistTrack>, Error> { 24 24 + let playlist_tracks: Vec<PlaylistTrack> = 25 25 + sqlx::query_as("SELECT * FROM playlist_tracks OFFSET $1 LIMIT $2") 26 26 + .bind(offset) 27 27 + .bind(limit) 28 28 + .fetch_all(pool) 29 29 + .await?; 30 30 + Ok(playlist_tracks) 31 31 + } 32 32 + 33 33 + pub async fn get_user_playlists( 34 34 + pool: &Pool<Postgres>, 35 35 + offset: i64, 36 36 + limit: i64, 37 37 + ) -> Result<Vec<UserPlaylist>, Error> { 38 38 + let user_playlists: Vec<UserPlaylist> = 39 39 + sqlx::query_as("SELECT * FROM user_playlists OFFSET $1 LIMIT $2") 40 40 + .bind(offset) 41 41 + .bind(limit) 42 42 + .fetch_all(pool) 43 43 + .await?; 44 44 + Ok(user_playlists) 45 45 + } 46 46 + 19 47 pub async fn insert_playlist(pool: &Pool<Postgres>, playlist: &Playlist) -> Result<(), Error> { 20 48 sqlx::query( 21 49 r#"INSERT INTO playlists ( ··· 48 76 .await?; 49 77 Ok(()) 50 78 } 79 79 + 80 80 + pub async fn insert_playlist_track( 81 81 + pool: &Pool<Postgres>, 82 82 + playlist_track: &PlaylistTrack, 83 83 + ) -> Result<(), Error> { 84 84 + sqlx::query( 85 85 + r#"INSERT INTO playlist_tracks ( 86 86 + xata_id, 87 87 + playlist_id, 88 88 + track_id, 89 89 + added_by, 90 90 + xata_createdat 91 91 + ) VALUES ($1, $2, $3, $4, $5) 92 92 + ON CONFLICT (xata_id) DO NOTHING"#, 93 93 + ) 94 94 + .bind(&playlist_track.xata_id) 95 95 + .bind(&playlist_track.playlist_id) 96 96 + .bind(&playlist_track.track_id) 97 97 + .bind(&playlist_track.added_by) 98 98 + .bind(playlist_track.xata_createdat) 99 99 + .execute(pool) 100 100 + .await?; 101 101 + Ok(()) 102 102 + } 103 103 + 104 104 + pub async fn insert_user_playlist( 105 105 + pool: &Pool<Postgres>, 106 106 + user_playlist: &UserPlaylist, 107 107 + ) -> Result<(), Error> { 108 108 + sqlx::query( 109 109 + r#"INSERT INTO user_playlists ( 110 110 + xata_id, 111 111 + user_id, 112 112 + playlist_id, 113 113 + uri, 114 114 + xata_createdat 115 115 + ) VALUES ($1, $2, $3, $4, $5) 116 116 + ON CONFLICT (xata_id) DO NOTHING"#, 117 117 + ) 118 118 + .bind(&user_playlist.xata_id) 119 119 + .bind(&user_playlist.user_id) 120 120 + .bind(&user_playlist.playlist_id) 121 121 + .bind(&user_playlist.uri) 122 122 + .bind(user_playlist.xata_createdat) 123 123 + .execute(pool) 124 124 + .await?; 125 125 + Ok(()) 126 126 + }

+37 -1

crates/pgpull/src/repo/track.rs

··· 1 1 use anyhow::Error; 2 2 use sqlx::{Pool, Postgres}; 3 3 4 4 - use crate::xata::track::Track; 4 4 + use crate::xata::{track::Track, user_track::UserTrack}; 5 5 6 6 pub async fn get_tracks( 7 7 pool: &Pool<Postgres>, ··· 14 14 .fetch_all(pool) 15 15 .await?; 16 16 Ok(tracks) 17 17 + } 18 18 + 19 19 + pub async fn get_user_tracks( 20 20 + pool: &Pool<Postgres>, 21 21 + offset: i64, 22 22 + limit: i64, 23 23 + ) -> Result<Vec<UserTrack>, Error> { 24 24 + let user_tracks: Vec<UserTrack> = 25 25 + sqlx::query_as("SELECT * FROM user_tracks OFFSET $1 LIMIT $2") 26 26 + .bind(offset) 27 27 + .bind(limit) 28 28 + .fetch_all(pool) 29 29 + .await?; 30 30 + Ok(user_tracks) 17 31 } 18 32 19 33 pub async fn insert_track(pool: &Pool<Postgres>, track: &Track) -> Result<(), Error> { ··· 75 89 .await?; 76 90 Ok(()) 77 91 } 92 92 + 93 93 + pub async fn insert_user_track(pool: &Pool<Postgres>, user_track: &UserTrack) -> Result<(), Error> { 94 94 + sqlx::query( 95 95 + r#"INSERT INTO user_tracks ( 96 96 + xata_id, 97 97 + user_id, 98 98 + track_id, 99 99 + uri, 100 100 + xata_createdat 101 101 + ) VALUES ($1, $2, $3, $4, $5) 102 102 + ON CONFLICT (xata_id) DO NOTHING 103 103 + "#, 104 104 + ) 105 105 + .bind(&user_track.xata_id) 106 106 + .bind(&user_track.user_id) 107 107 + .bind(&user_track.track_id) 108 108 + .bind(&user_track.uri) 109 109 + .bind(user_track.xata_createdat) 110 110 + .execute(pool) 111 111 + .await?; 112 112 + Ok(()) 113 113 + }

+1

crates/pgpull/src/xata/user_album.rs

··· 5 5 pub xata_id: String, 6 6 pub user_id: String, 7 7 pub album_id: String, 8 8 + pub uri: Option<String>, 8 9 #[serde(with = "chrono::serde::ts_seconds")] 9 10 pub xata_createdat: chrono::DateTime<chrono::Utc>, 10 11 }

+1

crates/pgpull/src/xata/user_artist.rs

··· 5 5 pub xata_id: String, 6 6 pub user_id: String, 7 7 pub artist_id: String, 8 8 + pub uri: Option<String>, 8 9 #[serde(with = "chrono::serde::ts_seconds")] 9 10 pub xata_createdat: chrono::DateTime<chrono::Utc>, 10 11 }

+1

crates/pgpull/src/xata/user_playlist.rs

··· 5 5 pub xata_id: String, 6 6 pub user_id: String, 7 7 pub playlist_id: String, 8 8 + pub uri: Option<String>, 8 9 #[serde(with = "chrono::serde::ts_seconds")] 9 10 pub xata_createdat: chrono::DateTime<chrono::Utc>, 10 11 }

+1

crates/pgpull/src/xata/user_track.rs

··· 5 5 pub xata_id: String, 6 6 pub user_id: String, 7 7 pub track_id: String, 8 8 + pub uri: Option<String>, 8 9 #[serde(with = "chrono::serde::ts_seconds")] 9 10 pub xata_createdat: chrono::DateTime<chrono::Utc>, 10 11 }