1use std::collections::{HashMap, HashSet};
2
3use color_eyre::eyre::Result;
4use sqlx::{Executor, Sqlite, SqlitePool};
5use tano_providers::local::parse_song::ParsedSong;
6
7use crate::{
8 album::{Album, CreateAlbum},
9 artist::{Artist, ArtistRole, CreateArtist},
10 builders::*,
11 bulk_builder::BulkBuilder,
12 local_song::{CreateLocalSong, LocalSong, SyncLocalSong},
13 song::{CreateSong, Song},
14};
15
16pub async fn get_songs(executor: impl Executor<'_, Database = Sqlite>) -> Result<Vec<Song>> {
17 let songs = sqlx::query_as!(
18 Song,
19 r#"
20 SELECT id, provider_id, album_id, title, track_number, duration, year
21 FROM songs
22 ORDER BY title
23 "#
24 )
25 .fetch_all(executor)
26 .await?;
27
28 Ok(songs)
29}
30
31pub async fn get_album_songs(
32 executor: impl Executor<'_, Database = Sqlite>,
33 album_id: i64,
34) -> Result<Vec<Song>> {
35 let songs = sqlx::query_as!(
36 Song,
37 r#"
38 SELECT id, provider_id, album_id, title, track_number, duration, year
39 FROM songs
40 WHERE album_id = ?
41 ORDER BY track_number, title
42 "#,
43 album_id
44 )
45 .fetch_all(executor)
46 .await?;
47
48 Ok(songs)
49}
50
51pub async fn get_album_artists(
52 executor: impl Executor<'_, Database = Sqlite>,
53 album_id: i64,
54) -> Result<Vec<Artist>> {
55 let artists = sqlx::query_as!(
56 Artist,
57 r#"
58 SELECT DISTINCT artists.id, artists.provider_id, artists.name
59 FROM artists
60 JOIN song_artists ON artists.id = song_artists.artist_id
61 JOIN songs ON song_artists.song_id = songs.id
62 WHERE songs.album_id = ? AND song_artists.role = 1
63 ORDER BY artists.name
64 "#,
65 album_id
66 )
67 .fetch_all(executor)
68 .await?;
69
70 Ok(artists)
71}
72
73pub async fn get_album(
74 executor: impl Executor<'_, Database = Sqlite>,
75 id: i64,
76) -> Result<Option<Album>> {
77 let album = sqlx::query_as!(
78 Album,
79 r#"
80 SELECT id, provider_id, title
81 FROM albums
82 WHERE id = ?
83 "#,
84 id
85 )
86 .fetch_optional(executor)
87 .await?;
88
89 Ok(album)
90}
91
92pub async fn get_albums(executor: impl Executor<'_, Database = Sqlite>) -> Result<Vec<Album>> {
93 let albums = sqlx::query_as!(
94 Album,
95 r#"
96 SELECT id, provider_id, title
97 FROM albums
98 ORDER BY title
99 "#
100 )
101 .fetch_all(executor)
102 .await?;
103
104 Ok(albums)
105}
106
107pub async fn get_artists(executor: impl Executor<'_, Database = Sqlite>) -> Result<Vec<Artist>> {
108 let artists = sqlx::query_as!(
109 Artist,
110 r#"
111 SELECT id, provider_id, name
112 FROM artists
113 ORDER BY name
114 "#
115 )
116 .fetch_all(executor)
117 .await?;
118
119 Ok(artists)
120}
121
122pub async fn get_song_ids(executor: impl Executor<'_, Database = Sqlite>) -> Result<Vec<i64>> {
123 let song_ids: Vec<i64> = sqlx::query_scalar!(
124 r#"
125 SELECT id
126 FROM songs
127 ORDER BY title
128 "#
129 )
130 .fetch_all(executor)
131 .await?;
132
133 Ok(song_ids)
134}
135
136pub async fn get_sync_local_songs(
137 executor: impl Executor<'_, Database = Sqlite>,
138 provider_id: u64,
139) -> Result<Vec<SyncLocalSong>> {
140 let records = sqlx::query_as!(
141 SyncLocalSong,
142 r#"
143 SELECT local_songs.song_id, local_songs.path, local_songs.inode, local_songs.mtime, local_songs.size
144 FROM local_songs
145 JOIN songs ON local_songs.song_id = songs.id
146 WHERE songs.provider_id = ?
147 "#,
148 provider_id as i64
149 )
150 .fetch_all(executor)
151 .await?;
152
153 Ok(records)
154}
155
156pub async fn get_local_song_by_path(
157 executor: impl Executor<'_, Database = Sqlite>,
158 provider_id: u64,
159 path: &str,
160) -> Result<Option<LocalSong>> {
161 let local_song = sqlx::query_as!(
162 LocalSong,
163 r#"
164 SELECT local_songs.song_id, local_songs.path, local_songs.inode, local_songs.mtime, local_songs.size, local_songs.format
165 FROM local_songs
166 JOIN songs ON local_songs.song_id = songs.id
167 WHERE local_songs.path = ? AND songs.provider_id = ?
168 LIMIT 1"#,
169 path,
170 provider_id as i64
171 )
172 .fetch_optional(executor)
173 .await?;
174
175 Ok(local_song)
176}
177
178pub async fn get_local_song_by_inode(
179 executor: impl Executor<'_, Database = Sqlite>,
180 provider_id: u64,
181 inode: i64,
182) -> Result<Option<LocalSong>> {
183 let local_song = sqlx::query_as!(
184 LocalSong,
185 r#"
186 SELECT local_songs.song_id, local_songs.path, local_songs.inode, local_songs.mtime, local_songs.size, local_songs.format
187 FROM local_songs
188 JOIN songs ON local_songs.song_id = songs.id
189 WHERE local_songs.inode = ? AND songs.provider_id = ?
190 LIMIT 1"#,
191 inode,
192 provider_id as i64
193 )
194 .fetch_optional(executor)
195 .await?;
196
197 Ok(local_song)
198}
199
200pub async fn insert_song(
201 executor: impl Executor<'_, Database = Sqlite>,
202 song: &CreateSong,
203) -> Result<i64> {
204 let record = sqlx::query!(
205 r#"
206 INSERT INTO songs (provider_id, album_id, title, track_number, duration, year)
207 VALUES (?, ?, ?, ?, ?, ?)
208 RETURNING id AS "id!"
209 "#,
210 song.provider_id,
211 song.album_id,
212 song.title,
213 song.track_number,
214 song.duration,
215 song.year
216 )
217 .fetch_one(executor)
218 .await?;
219
220 Ok(record.id)
221}
222
223pub async fn insert_local_song(
224 executor: impl Executor<'_, Database = Sqlite>,
225 song_id: i64,
226 local_song: &CreateLocalSong,
227) -> Result<()> {
228 sqlx::query!(
229 r#"
230 INSERT INTO local_songs (song_id, path, inode, mtime, size, format)
231 VALUES (?, ?, ?, ?, ?, ?)
232 "#,
233 song_id,
234 local_song.path,
235 local_song.inode,
236 local_song.mtime,
237 local_song.size,
238 local_song.format
239 )
240 .execute(executor)
241 .await?;
242
243 Ok(())
244}
245
246pub async fn update_song(
247 executor: impl Executor<'_, Database = Sqlite>,
248 id: i64,
249 song: &CreateSong,
250) -> Result<()> {
251 sqlx::query!(
252 r#"
253 UPDATE songs
254 SET title = ?, provider_id = ?, album_id = ?, track_number = ?, duration = ?, year = ?
255 WHERE id = ?
256 "#,
257 song.title,
258 song.provider_id,
259 song.album_id,
260 song.track_number,
261 song.duration,
262 song.year,
263 id
264 )
265 .execute(executor)
266 .await?;
267
268 Ok(())
269}
270
271pub async fn update_local_song(
272 executor: impl Executor<'_, Database = Sqlite>,
273 id: i64,
274 local_song: &CreateLocalSong,
275) -> Result<()> {
276 sqlx::query!(
277 r#"
278 UPDATE local_songs
279 SET path = ?, inode = ?, mtime = ?, size = ?, format = ?
280 WHERE song_id = ?
281 "#,
282 local_song.path,
283 local_song.inode,
284 local_song.mtime,
285 local_song.size,
286 local_song.format,
287 id
288 )
289 .execute(executor)
290 .await?;
291
292 Ok(())
293}
294
295pub async fn update_local_song_path(
296 executor: impl Executor<'_, Database = Sqlite>,
297 id: i64,
298 path: &str,
299) -> Result<()> {
300 sqlx::query!(
301 r#"
302 UPDATE local_songs
303 SET path = ?
304 WHERE song_id = ?
305 "#,
306 path,
307 id
308 )
309 .execute(executor)
310 .await?;
311
312 Ok(())
313}
314
315pub async fn delete_song(executor: impl Executor<'_, Database = Sqlite>, id: i64) -> Result<()> {
316 sqlx::query!(
317 r#"
318 DELETE FROM songs WHERE id = ?
319 "#,
320 id
321 )
322 .execute(executor)
323 .await?;
324
325 Ok(())
326}
327
328pub async fn upsert_album(
329 executor: impl Executor<'_, Database = Sqlite>,
330 album: &CreateAlbum,
331) -> Result<i64> {
332 let record = sqlx::query!(
333 r#"
334 INSERT INTO albums (provider_id, title)
335 VALUES (?, ?)
336 ON CONFLICT (provider_id, title) DO UPDATE SET title = excluded.title
337 RETURNING id AS "id!"
338 "#,
339 album.provider_id,
340 album.title
341 )
342 .fetch_one(executor)
343 .await?;
344
345 Ok(record.id)
346}
347
348pub async fn delete_orphan_albums(executor: impl Executor<'_, Database = Sqlite>) -> Result<()> {
349 sqlx::query!(
350 r#"
351 DELETE FROM albums
352 WHERE id NOT IN (SELECT album_id FROM songs)
353 "#
354 )
355 .execute(executor)
356 .await?;
357
358 Ok(())
359}
360
361pub async fn upsert_artist(
362 executor: impl Executor<'_, Database = Sqlite>,
363 artist: &CreateArtist,
364) -> Result<i64> {
365 let record = sqlx::query!(
366 r#"
367 INSERT INTO artists (provider_id, name)
368 VALUES (?, ?)
369 ON CONFLICT (provider_id, name) DO UPDATE SET name = excluded.name
370 RETURNING id AS "id!"
371 "#,
372 artist.provider_id,
373 artist.name
374 )
375 .fetch_one(executor)
376 .await?;
377
378 Ok(record.id)
379}
380
381pub async fn delete_orphan_artists(executor: impl Executor<'_, Database = Sqlite>) -> Result<()> {
382 sqlx::query!(
383 r#"
384 DELETE FROM artists
385 WHERE id NOT IN (SELECT artist_id FROM song_artists)
386 "#
387 )
388 .execute(executor)
389 .await?;
390
391 Ok(())
392}
393
394pub async fn insert_song_artist(
395 executor: impl Executor<'_, Database = Sqlite>,
396 song_id: i64,
397 artist_id: i64,
398 role: ArtistRole,
399) -> Result<()> {
400 let role_id = role as i64;
401 sqlx::query!(
402 r#"
403 INSERT OR IGNORE INTO song_artists (song_id, artist_id, role)
404 VALUES (?, ?, ?)
405 "#,
406 song_id,
407 artist_id,
408 role_id
409 )
410 .execute(executor)
411 .await?;
412
413 Ok(())
414}
415
416pub async fn delete_song_artists(
417 executor: impl Executor<'_, Database = Sqlite>,
418 song_id: i64,
419) -> Result<()> {
420 sqlx::query!(
421 r#"
422 DELETE FROM song_artists WHERE song_id = ?
423 "#,
424 song_id
425 )
426 .execute(executor)
427 .await?;
428
429 Ok(())
430}
431
432pub async fn sync_local_songs(
433 pool: &SqlitePool,
434 provider_id: u64,
435 new_songs: Vec<ParsedSong>,
436 updated_songs: Vec<(i64, ParsedSong)>,
437 to_update_path: Vec<(i64, String)>,
438 to_delete_ids: Vec<i64>,
439) -> Result<()> {
440 let mut tx = pool.begin().await?;
441 let chunk_size = 1000;
442
443 let mut unique_artists = HashSet::new();
444 for parsed in new_songs
445 .iter()
446 .chain(updated_songs.iter().map(|(_, song)| song))
447 {
448 unique_artists.insert(parsed.album_artist_name.as_str());
449 unique_artists.insert(parsed.artist_name.as_str());
450 }
451
452 let provider_id = provider_id as i64;
453
454 let mut artist_map = HashMap::new();
455 for chunk in unique_artists.iter().collect::<Vec<_>>().chunks(chunk_size) {
456 let mut builder = UpsertArtistsBuilder::new();
457 for &&name in chunk {
458 let artist = CreateArtist {
459 provider_id,
460 name: name.to_string(),
461 };
462 builder.push(&artist);
463 }
464
465 let records = builder
466 .build()
467 .build_query_as::<(i64, String)>()
468 .fetch_all(&mut *tx)
469 .await?;
470
471 for (id, db_name) in records {
472 if let Some(&orig_name) = unique_artists.get(db_name.as_str()) {
473 artist_map.insert(orig_name, id);
474 }
475 }
476 }
477
478 let mut unique_albums = HashSet::new();
479 for parsed in new_songs
480 .iter()
481 .chain(updated_songs.iter().map(|(_, song)| song))
482 {
483 unique_albums.insert(parsed.album_title.as_str());
484 }
485
486 let mut album_map = HashMap::new();
487 for chunk in unique_albums.iter().collect::<Vec<_>>().chunks(chunk_size) {
488 let mut builder = UpsertAlbumsBuilder::new();
489 for &&title in chunk {
490 let album = CreateAlbum {
491 provider_id,
492 title: title.to_string(),
493 };
494 builder.push(&album);
495 }
496
497 let records = builder
498 .build()
499 .build_query_as::<(i64, String)>()
500 .fetch_all(&mut *tx)
501 .await?;
502
503 for (id, db_title) in records {
504 if let Some(&orig_title) = unique_albums.get(db_title.as_str()) {
505 album_map.insert(orig_title, id);
506 }
507 }
508 }
509
510 let mut pending_song_artists: Vec<(i64, i64, ArtistRole)> = Vec::new();
511
512 for chunk in new_songs.chunks(chunk_size) {
513 let mut builder = InsertSongsBuilder::new();
514 for parsed in chunk {
515 let album_id = *album_map.get(parsed.album_title.as_str()).unwrap();
516 let song = CreateSong {
517 provider_id,
518 album_id,
519 title: parsed.title.clone(),
520 track_number: parsed.track_number,
521 duration: parsed.duration,
522 year: parsed.year,
523 };
524 builder.push(&song);
525 }
526
527 let returned_ids = builder
528 .build()
529 .build_query_as::<(i64,)>()
530 .fetch_all(&mut *tx)
531 .await?;
532
533 let mut local_builder = InsertLocalSongsBuilder::new();
534 for (parsed, (song_id,)) in chunk.iter().zip(returned_ids.iter()) {
535 let local_song = CreateLocalSong {
536 path: parsed.path.clone(),
537 inode: parsed.inode,
538 mtime: parsed.mtime,
539 size: parsed.size,
540 format: parsed.format.clone(),
541 };
542 local_builder.push((*song_id, &local_song));
543
544 let album_artist_id = *artist_map.get(parsed.album_artist_name.as_str()).unwrap();
545 let track_artist_id = *artist_map.get(parsed.artist_name.as_str()).unwrap();
546 pending_song_artists.push((*song_id, album_artist_id, ArtistRole::AlbumArtist));
547 pending_song_artists.push((*song_id, track_artist_id, ArtistRole::Artist));
548 }
549 local_builder.build().build().execute(&mut *tx).await?;
550 }
551
552 for chunk in updated_songs.chunks(chunk_size) {
553 let mut builder = UpdateSongsBuilder::new();
554 for (id, parsed) in chunk {
555 let album_id = *album_map.get(parsed.album_title.as_str()).unwrap();
556 let song = CreateSong {
557 provider_id,
558 album_id,
559 title: parsed.title.clone(),
560 track_number: parsed.track_number,
561 duration: parsed.duration,
562 year: parsed.year,
563 };
564 builder.push((*id, &song));
565 }
566 builder.build().build().execute(&mut *tx).await?;
567
568 let mut local_builder = UpdateLocalSongsBuilder::new();
569 for (id, parsed) in chunk {
570 let local_song = CreateLocalSong {
571 path: parsed.path.clone(),
572 inode: parsed.inode,
573 mtime: parsed.mtime,
574 size: parsed.size,
575 format: parsed.format.clone(),
576 };
577 local_builder.push((*id, &local_song));
578
579 let album_artist_id = *artist_map.get(parsed.album_artist_name.as_str()).unwrap();
580 let track_artist_id = *artist_map.get(parsed.artist_name.as_str()).unwrap();
581 pending_song_artists.push((*id, album_artist_id, ArtistRole::AlbumArtist));
582 pending_song_artists.push((*id, track_artist_id, ArtistRole::Artist));
583 }
584 local_builder.build().build().execute(&mut *tx).await?;
585
586 let mut delete_builder = DeleteSongArtistsBuilder::new();
587 for (id, _) in chunk {
588 delete_builder.push(*id);
589 }
590 delete_builder.build().build().execute(&mut *tx).await?;
591 }
592
593 for chunk in pending_song_artists.chunks(chunk_size) {
594 let mut builder = InsertSongArtistsBuilder::new();
595 for &(song_id, artist_id, role) in chunk {
596 builder.push((song_id, artist_id, role));
597 }
598 builder.build().build().execute(&mut *tx).await?;
599 }
600
601 for chunk in to_update_path.chunks(chunk_size) {
602 let mut builder = UpdateLocalSongsPathBuilder::new();
603 for (id, path) in chunk {
604 builder.push((*id, path.as_str()));
605 }
606 builder.build().build().execute(&mut *tx).await?;
607 }
608
609 for chunk in to_delete_ids.chunks(chunk_size) {
610 let mut builder = DeleteSongsBuilder::new();
611 for &id in chunk {
612 builder.push(id);
613 }
614 builder.build().build().execute(&mut *tx).await?;
615 }
616
617 delete_orphan_albums(&mut *tx).await?;
618 delete_orphan_artists(&mut *tx).await?;
619
620 tx.commit().await?;
621
622 Ok(())
623}
624
625pub async fn insert_parsed_song(
626 pool: &SqlitePool,
627 provider_id: u64,
628 parsed: &ParsedSong,
629) -> Result<i64> {
630 let mut tx = pool.begin().await?;
631
632 let album_id = upsert_album(
633 &mut *tx,
634 &CreateAlbum {
635 provider_id: provider_id as i64,
636 title: parsed.album_title.clone(),
637 },
638 )
639 .await?;
640
641 let song_id = insert_song(
642 &mut *tx,
643 &CreateSong {
644 provider_id: provider_id as i64,
645 album_id,
646 title: parsed.title.clone(),
647 track_number: parsed.track_number,
648 duration: parsed.duration,
649 year: parsed.year,
650 },
651 )
652 .await?;
653
654 let album_artist_id = upsert_artist(
655 &mut *tx,
656 &CreateArtist {
657 provider_id: provider_id as i64,
658 name: parsed.album_artist_name.clone(),
659 },
660 )
661 .await?;
662 insert_song_artist(&mut *tx, song_id, album_artist_id, ArtistRole::AlbumArtist).await?;
663
664 insert_local_song(
665 &mut *tx,
666 song_id,
667 &CreateLocalSong {
668 path: parsed.path.clone(),
669 inode: parsed.inode,
670 mtime: parsed.mtime,
671 size: parsed.size,
672 format: parsed.format.clone(),
673 },
674 )
675 .await?;
676
677 let artist_id = upsert_artist(
678 &mut *tx,
679 &CreateArtist {
680 provider_id: provider_id as i64,
681 name: parsed.artist_name.clone(),
682 },
683 )
684 .await?;
685 insert_song_artist(&mut *tx, song_id, artist_id, ArtistRole::Artist).await?;
686
687 tx.commit().await?;
688
689 Ok(song_id)
690}
691
692pub async fn update_parsed_song(
693 pool: &SqlitePool,
694 provider_id: u64,
695 id: i64,
696 parsed: &ParsedSong,
697) -> Result<()> {
698 let mut tx = pool.begin().await?;
699
700 let album_id = upsert_album(
701 &mut *tx,
702 &CreateAlbum {
703 provider_id: provider_id as i64,
704 title: parsed.album_title.clone(),
705 },
706 )
707 .await?;
708
709 update_song(
710 &mut *tx,
711 id,
712 &CreateSong {
713 provider_id: provider_id as i64,
714 album_id,
715 title: parsed.title.clone(),
716 track_number: parsed.track_number,
717 duration: parsed.duration,
718 year: parsed.year,
719 },
720 )
721 .await?;
722
723 update_local_song(
724 &mut *tx,
725 id,
726 &CreateLocalSong {
727 path: parsed.path.clone(),
728 inode: parsed.inode,
729 mtime: parsed.mtime,
730 size: parsed.size,
731 format: parsed.format.clone(),
732 },
733 )
734 .await?;
735
736 delete_song_artists(&mut *tx, id).await?;
737
738 let album_artist_id = upsert_artist(
739 &mut *tx,
740 &CreateArtist {
741 provider_id: provider_id as i64,
742 name: parsed.album_artist_name.clone(),
743 },
744 )
745 .await?;
746 insert_song_artist(&mut *tx, id, album_artist_id, ArtistRole::AlbumArtist).await?;
747
748 let artist_id = upsert_artist(
749 &mut *tx,
750 &CreateArtist {
751 provider_id: provider_id as i64,
752 name: parsed.artist_name.clone(),
753 },
754 )
755 .await?;
756 insert_song_artist(&mut *tx, id, artist_id, ArtistRole::Artist).await?;
757
758 delete_orphan_albums(&mut *tx).await?;
759 delete_orphan_artists(&mut *tx).await?;
760
761 tx.commit().await?;
762
763 Ok(())
764}
765
766pub async fn delete_parsed_song(pool: &SqlitePool, id: i64) -> Result<()> {
767 let mut tx = pool.begin().await?;
768
769 delete_song(&mut *tx, id).await?;
770 delete_orphan_albums(&mut *tx).await?;
771 delete_orphan_artists(&mut *tx).await?;
772
773 tx.commit().await?;
774
775 Ok(())
776}