Apply the collision policy to exports bound for the server

A queued export could not see the server, so its name check always
answered "free" and the upload PUT over whatever was there: Increment
and Skip behaved as Overwrite on Nextcloud, and two exports of the same
name queued before either uploaded landed on one file.

The batch now names around what the album records of earlier exports
and what the outbox already holds for that folder. The outbox record
carries the policy, and the drain lists each destination folder once
and applies it against what the server holds: Increment steps past a
taken name and re-points the album's row, Skip drops the entry. A
record without a policy (older builds, a merge's composite) is sent as
named, as before. The album is recorded before the drain starts so a
rename has a row to move.

core/dr-catalog/src/albums.rs

@@ -306,6 +306,48 @@ pub fn record_exports(
     Ok(())
 }
 
+/// Every file name an album records, for an export choosing a name to know
+/// what it would land on.
+///
+/// A server album cannot be asked while the export is queued offline, and
+/// the names this app put there are the ones a second export of the same
+/// photographs will collide with. One read of the album's rows, not one per
+/// candidate name.
+pub fn file_names(
+    conn: &Connection,
+    id: AlbumId,
+) -> Result<std::collections::HashSet<String>, CatalogError> {
+    ensure_tables(conn)?;
+    let mut stmt = conn.prepare("SELECT file_name FROM album_exports WHERE album_id = ?1")?;
+    let rows = stmt
+        .query_map([id.0 as i64], |r| r.get(0))?
+        .collect::<Result<_, _>>()?;
+    Ok(rows)
+}
+
+/// A file the upload had to give another name: the server held one by the
+/// name the export recorded, put there by something this catalog never
+/// saw. The album row follows the file to the name it was given.
+///
+/// By the album's server folder, because that is all an outbox entry knows.
+/// `folder` is spelled as [`Place::Server`] spells it, without slashes at
+/// either end.
+pub fn rename_export(
+    conn: &Connection,
+    folder: &str,
+    from: &str,
+    to: &str,
+) -> Result<(), CatalogError> {
+    ensure_tables(conn)?;
+    conn.execute(
+        "UPDATE OR REPLACE album_exports SET file_name = ?3
+          WHERE file_name = ?2
+            AND album_id IN (SELECT id FROM albums WHERE server_path = ?1 AND deleted = 0)",
+        rusqlite::params![folder.trim_matches('/'), from, to],
+    )?;
+    Ok(())
+}
+
 /// The photographs behind an album's files, most recently exported first —
 /// what the grid shows when the album is opened.
 pub fn sources(conn: &Connection, id: AlbumId) -> Result<Vec<ImageId>, CatalogError> {
@@ -463,6 +505,28 @@ mod tests {
         assert_eq!(sources(conn, album).unwrap(), vec![b]);
     }
 
+    #[test]
+    fn a_renamed_upload_moves_the_row_of_the_server_album_only() {
+        let cat = catalog();
+        let conn = cat.connection();
+        let web = create(conn, "Web", &Place::Server("Albums/Web".into())).unwrap();
+        let other = create(conn, "Other", &Place::Server("Albums/Other".into())).unwrap();
+        let a = image(conn, "a.cr3");
+        record_exports(conn, web, &[(a, "a.jpg".into())]).unwrap();
+        record_exports(conn, other, &[(a, "a.jpg".into())]).unwrap();
+
+        rename_export(conn, "/Albums/Web", "a.jpg", "a-1.jpg").unwrap();
+
+        assert_eq!(
+            file_names(conn, web).unwrap(),
+            ["a-1.jpg".to_string()].into()
+        );
+        assert_eq!(
+            file_names(conn, other).unwrap(),
+            ["a.jpg".to_string()].into()
+        );
+    }
+
     #[test]
     fn moving_to_the_server_forgets_the_local_folder() {
         let cat = catalog();