refactor: move bulk_ops to queries and scanner service

- Delete bulk_ops.py, moving logic to appropriate layers
- Add bulk insert query functions:
  - queries/asset.bulk_insert_assets
  - queries/cache_state.bulk_insert_cache_states_ignore_conflicts
  - queries/cache_state.get_cache_states_by_paths_and_asset_ids
  - queries/asset_info.bulk_insert_asset_infos_ignore_conflicts
  - queries/asset_info.get_asset_info_ids_by_ids
  - queries/tags.bulk_insert_tags_and_meta
- Move seed_from_paths_batch orchestration to scanner._seed_from_paths_batch

Amp-Thread-ID: https://ampcode.com/threads/T-019c24fd-157d-776a-ad24-4f19cf5d3afe
Co-authored-by: Amp <amp@ampcode.com>

app/assets/database/queries/cache_state.py

@@ -9,6 +9,8 @@ from sqlalchemy.orm import Session
 from app.assets.database.models import Asset, AssetCacheState, AssetInfo
 from app.assets.helpers import escape_like_prefix
 
+MAX_BIND_PARAMS = 800
+
 __all__ = [
     "CacheStateRow",
     "list_cache_states_by_asset_id",
@@ -20,9 +22,20 @@ __all__ = [
     "bulk_set_needs_verify",
     "delete_cache_states_by_ids",
     "delete_orphaned_seed_asset",
+    "bulk_insert_cache_states_ignore_conflicts",
+    "get_cache_states_by_paths_and_asset_ids",
 ]
 
 
+def _rows_per_stmt(cols: int) -> int:
+    return max(1, MAX_BIND_PARAMS // max(1, cols))
+
+
+def _iter_chunks(seq, n: int):
+    for i in range(0, len(seq), n):
+        yield seq[i : i + n]
+
+
 class CacheStateRow(NamedTuple):
     """Row from cache state query with joined asset data."""
 
@@ -233,3 +246,50 @@ def delete_orphaned_seed_asset(session: Session, asset_id: str) -> bool:
         session.delete(asset)
         return True
     return False
+
+
+def bulk_insert_cache_states_ignore_conflicts(
+    session: Session,
+    rows: list[dict],
+) -> None:
+    """Bulk insert cache state rows with ON CONFLICT DO NOTHING on file_path.
+
+    Each dict should have: asset_id, file_path, mtime_ns
+    """
+    if not rows:
+        return
+    ins = sqlite.insert(AssetCacheState).on_conflict_do_nothing(
+        index_elements=[AssetCacheState.file_path]
+    )
+    for chunk in _iter_chunks(rows, _rows_per_stmt(3)):
+        session.execute(ins, chunk)
+
+
+def get_cache_states_by_paths_and_asset_ids(
+    session: Session,
+    path_to_asset: dict[str, str],
+) -> set[str]:
+    """Query cache states to find paths where our asset_id won the insert.
+
+    Args:
+        path_to_asset: Mapping of file_path -> asset_id we tried to insert
+
+    Returns:
+        Set of file_paths where our asset_id is present
+    """
+    if not path_to_asset:
+        return set()
+
+    paths = list(path_to_asset.keys())
+    winners: set[str] = set()
+
+    for chunk in _iter_chunks(paths, MAX_BIND_PARAMS):
+        result = session.execute(
+            select(AssetCacheState.file_path).where(
+                AssetCacheState.file_path.in_(chunk),
+                AssetCacheState.asset_id.in_([path_to_asset[p] for p in chunk]),
+            )
+        )
+        winners.update(result.scalars().all())
+
+    return winners