feat(modules): group-files schema + owner-relative path helpers + repo (§11.6 files C1)

Extends the §11.6 shared-collection machinery to the filesystem-backed
`files` blob store (after KV/0053 and docs/0054).

- 0055_group_files.sql: widen the group_collections kind CHECK to admit
  'files'; add a group-keyed `group_files` metadata table mirroring
  `files` (0018), CASCADE on group delete.
- files_repo: generalize the four path/IO free functions
  (shard_dir_at / final_path_at / write_atomic_at / read_verify_at) to
  take an owner-relative dir instead of `app_id`, and make them (plus the
  cursor helpers) pub(crate). The security-sensitive atomic-write +
  checksum-on-read mechanics now have a single source shared by the app
  and group repos. App behavior is unchanged (apps shard at `<app_id>/`).
- group_files_repo: FsGroupFilesRepo keyed by group_id, blobs at
  `<root>/files/groups/<group_id>/<collection>/<id[0:2]>/<id>` — a
  `groups/` infix disjoint from the per-app subtree, so the existing
  recursive orphan sweeper covers it with zero change.

Schema snapshot blessed (55 migrations); app-files fs tests still green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/manager-core/migrations/0055_group_files.sql

@@ -0,0 +1,37 @@
+-- §11.6 (cont., v1.2 Hierarchies): group-shared FILES collections.
+--
+-- Extends the shared-collection machinery (0052 registry + the per-kind stores
+-- 0053 group_kv_entries / 0054 group_docs) from KV/docs to the filesystem-backed
+-- `files` blob store. The registry's `kind` discriminator was built for exactly
+-- this — widen its CHECK to admit 'files', and add a group-keyed metadata table
+-- mirroring `files` (0018).
+
+-- Widen the marker kind allow-list (auto-named `group_collections_kind_check`,
+-- per 0052's inline column CHECK; 0054 widened it to ('kv','docs')).
+ALTER TABLE group_collections
+    DROP CONSTRAINT group_collections_kind_check,
+    ADD CONSTRAINT group_collections_kind_check CHECK (kind IN ('kv', 'docs', 'files'));
+
+-- Group-keyed file metadata. Identity tuple `(group_id, collection, id)` — keyed
+-- by the OWNING GROUP, not an app (the shared blobs belong to the group). The
+-- bytes live on disk at
+--   <PICLOUD_FILES_ROOT>/files/groups/<group_id>/<collection>/<id[0:2]>/<id>
+-- (a `groups/` infix disjoint from the per-app `files/<app_id>/...` subtree, so
+-- the orphan sweeper covers both with one walk). CASCADE on group delete (the
+-- metadata dies with its group; an app delete leaves it), matching 0053/0054.
+CREATE TABLE group_files (
+    group_id        UUID NOT NULL REFERENCES groups(id) ON DELETE CASCADE,
+    collection      TEXT NOT NULL,
+    id              UUID NOT NULL,
+    name            TEXT NOT NULL,
+    content_type    TEXT NOT NULL,
+    size_bytes      BIGINT NOT NULL,
+    checksum_sha256 TEXT NOT NULL,           -- hex, 64 chars, lowercase
+    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    PRIMARY KEY (group_id, collection, id)
+);
+
+-- List + cursor pagination scans by (group_id, collection) — mirrors
+-- idx_files_app_collection (0018).
+CREATE INDEX idx_group_files_group_collection ON group_files (group_id, collection);

crates/manager-core/src/files_repo.rs

@@ -211,7 +211,7 @@ impl FsFilesRepo {
     }
 
     fn final_path(&self, app_id: AppId, collection: &str, id: Uuid) -> PathBuf {
-        final_path_at(&self.config.root, app_id, collection, id)
+        final_path_at(&self.config.root, &app_owner_dir(app_id), collection, id)
     }
 
     fn write_atomic(
@@ -221,29 +221,58 @@ impl FsFilesRepo {
         id: Uuid,
         bytes: &[u8],
     ) -> Result<String, FilesRepoError> {
-        write_atomic_at(&self.config.root, app_id, collection, id, bytes)
+        write_atomic_at(
+            &self.config.root,
+            &app_owner_dir(app_id),
+            collection,
+            id,
+            bytes,
+        )
     }
 }
 
-fn shard_dir_at(root: &Path, app_id: AppId, collection: &str, id_str: &str) -> PathBuf {
+/// The owner-relative subdirectory of an **app**'s blobs under `<root>/files/`:
+/// just `<app_id>`. Group-shared blobs (§11.6) shard at `groups/<group_id>`
+/// instead (see `group_files_repo::group_owner_dir`) — a disjoint subtree under
+/// the same `files/` base, so the orphan sweeper covers both with one walk.
+pub(crate) fn app_owner_dir(app_id: AppId) -> PathBuf {
+    PathBuf::from(app_id.into_inner().to_string())
+}
+
+/// `<root>/files/<owner_rel>/<collection>/<id[0:2]>/`. `owner_rel` is built
+/// by the caller from a server-generated id (`<app_id>` or `groups/<group_id>`)
+/// — never user input — and `collection` is path-guarded one layer up, so no
+/// component can escape the `files/` base.
+pub(crate) fn shard_dir_at(
+    root: &Path,
+    owner_rel: &Path,
+    collection: &str,
+    id_str: &str,
+) -> PathBuf {
     root.join("files")
-        .join(app_id.into_inner().to_string())
+        .join(owner_rel)
         .join(collection)
         .join(&id_str[..2])
 }
 
-fn final_path_at(root: &Path, app_id: AppId, collection: &str, id: Uuid) -> PathBuf {
+pub(crate) fn final_path_at(
+    root: &Path,
+    owner_rel: &Path,
+    collection: &str,
+    id: Uuid,
+) -> PathBuf {
     let id_str = id.to_string();
-    shard_dir_at(root, app_id, collection, &id_str).join(&id_str)
+    shard_dir_at(root, owner_rel, collection, &id_str).join(&id_str)
 }
 
 /// Steps 2–6 of the atomic-write protocol. Returns the lowercase hex
 /// SHA-256 of the bytes (computed in a single pass over the in-memory
-/// buffer — the file is never re-read). Free function so the fs
+/// buffer — the file is never re-read). `pub(crate)` + owner-relative so the
-/// mechanics are unit-testable without a Postgres pool.
+/// app and group-shared (§11.6) repos share this single source for the
-fn write_atomic_at(
+/// security-sensitive write+checksum mechanics, unit-testable without a pool.
+pub(crate) fn write_atomic_at(
     root: &Path,
-    app_id: AppId,
+    owner_rel: &Path,
     collection: &str,
     id: Uuid,
     bytes: &[u8],
@@ -251,7 +280,7 @@ fn write_atomic_at(
     use std::io::Write as _;
 
     let id_str = id.to_string();
-    let dir = shard_dir_at(root, app_id, collection, &id_str);
+    let dir = shard_dir_at(root, owner_rel, collection, &id_str);
     create_dir_all_secure(&dir)?;
 
     // Single-pass checksum over the in-memory buffer.
@@ -278,15 +307,16 @@ fn write_atomic_at(
 
 /// Read + checksum-verify the bytes at the given path-set. Free
 /// function mirror of the `get` read path. Returns `Corrupted` when the
-/// bytes are missing or don't match `expected_checksum`.
+/// bytes are missing or don't match `expected_checksum`. `pub(crate)` +
-fn read_verify_at(
+/// owner-relative — shared with the group-files (§11.6) repo.
+pub(crate) fn read_verify_at(
     root: &Path,
-    app_id: AppId,
+    owner_rel: &Path,
     collection: &str,
     id: Uuid,
     expected_checksum: &str,
 ) -> Result<Vec<u8>, FilesRepoError> {
-    let path = final_path_at(root, app_id, collection, id);
+    let path = final_path_at(root, owner_rel, collection, id);
     let bytes = match std::fs::read(&path) {
         Ok(b) => b,
         Err(e) => {
@@ -383,7 +413,13 @@ impl FilesRepo for FsFilesRepo {
         let Some((stored_checksum,)) = row else {
             return Ok(None);
         };
-        let bytes = read_verify_at(&self.config.root, app_id, collection, id, &stored_checksum)?;
+        let bytes = read_verify_at(
+            &self.config.root,
+            &app_owner_dir(app_id),
+            collection,
+            id,
+            &stored_checksum,
+        )?;
         Ok(Some(bytes))
     }
 
@@ -555,11 +591,11 @@ fn hex_lower(bytes: &[u8]) -> String {
     s
 }
 
-fn encode_cursor(last_id: Uuid) -> String {
+pub(crate) fn encode_cursor(last_id: Uuid) -> String {
     URL_SAFE_NO_PAD.encode(last_id.to_string().as_bytes())
 }
 
-fn decode_cursor(cursor: &str) -> Result<Uuid, FilesRepoError> {
+pub(crate) fn decode_cursor(cursor: &str) -> Result<Uuid, FilesRepoError> {
     let bytes = URL_SAFE_NO_PAD
         .decode(cursor)
         .map_err(|_| FilesRepoError::InvalidCursor)?;
@@ -672,13 +708,13 @@ mod tests {
         let id = Uuid::new_v4();
         let bytes = b"hello picloud files".to_vec();
 
-        let checksum = write_atomic_at(&root, app, "avatars", id, &bytes).unwrap();
+        let checksum = write_atomic_at(&root, &app_owner_dir(app), "avatars", id, &bytes).unwrap();
         // Single-pass checksum matches an independent hash of the bytes.
         let mut h = Sha256::new();
         h.update(&bytes);
         assert_eq!(checksum, hex_lower(&h.finalize()));
 
-        let read = read_verify_at(&root, app, "avatars", id, &checksum).unwrap();
+        let read = read_verify_at(&root, &app_owner_dir(app), "avatars", id, &checksum).unwrap();
         assert_eq!(read, bytes);
 
         std::fs::remove_dir_all(&root).ok();
@@ -689,13 +725,13 @@ mod tests {
         let root = unique_tmp_root();
         let app = AppId::new();
         let id = Uuid::new_v4();
-        let checksum = write_atomic_at(&root, app, "c", id, b"original").unwrap();
+        let checksum = write_atomic_at(&root, &app_owner_dir(app), "c", id, b"original").unwrap();
 
         // Mutate the bytes behind the repo's back.
-        let path = final_path_at(&root, app, "c", id);
+        let path = final_path_at(&root, &app_owner_dir(app), "c", id);
         std::fs::write(&path, b"tampered").unwrap();
 
-        let err = read_verify_at(&root, app, "c", id, &checksum).unwrap_err();
+        let err = read_verify_at(&root, &app_owner_dir(app), "c", id, &checksum).unwrap_err();
         assert!(matches!(err, FilesRepoError::Corrupted));
 
         std::fs::remove_dir_all(&root).ok();
@@ -707,7 +743,7 @@ mod tests {
         let app = AppId::new();
         let id = Uuid::new_v4();
         // No write — the file never existed.
-        let err = read_verify_at(&root, app, "c", id, "deadbeef").unwrap_err();
+        let err = read_verify_at(&root, &app_owner_dir(app), "c", id, "deadbeef").unwrap_err();
         assert!(matches!(err, FilesRepoError::Corrupted));
         std::fs::remove_dir_all(&root).ok();
     }
@@ -717,10 +753,10 @@ mod tests {
         let root = unique_tmp_root();
         let app = AppId::new();
         let id = Uuid::new_v4();
-        write_atomic_at(&root, app, "c", id, b"data").unwrap();
+        write_atomic_at(&root, &app_owner_dir(app), "c", id, b"data").unwrap();
 
         let id_str = id.to_string();
-        let dir = shard_dir_at(&root, app, "c", &id_str);
+        let dir = shard_dir_at(&root, &app_owner_dir(app), "c", &id_str);
         let entries: Vec<_> = std::fs::read_dir(&dir)
             .unwrap()
             .filter_map(Result::ok)
@@ -739,7 +775,7 @@ mod tests {
         let app = AppId::new();
         let id = Uuid::new_v4();
         let id_str = id.to_string();
-        let path = final_path_at(&root, app, "col", id);
+        let path = final_path_at(&root, &app_owner_dir(app), "col", id);
         let shard = &id_str[..2];
         assert!(path
             .to_string_lossy()
@@ -753,9 +789,9 @@ mod tests {
         let root = unique_tmp_root();
         let app = AppId::new();
         let id = Uuid::new_v4();
-        write_atomic_at(&root, app, "c", id, b"data").unwrap();
+        write_atomic_at(&root, &app_owner_dir(app), "c", id, b"data").unwrap();
         let id_str = id.to_string();
-        let dir = shard_dir_at(&root, app, "c", &id_str);
+        let dir = shard_dir_at(&root, &app_owner_dir(app), "c", &id_str);
         let mode = std::fs::metadata(&dir).unwrap().permissions().mode();
         assert_eq!(mode & 0o777, 0o700, "shard dir should be 0o700");
         std::fs::remove_dir_all(&root).ok();

crates/manager-core/src/group_files_repo.rs

@@ -0,0 +1,399 @@
+//! Low-level metadata (Postgres `group_files`) + blob bytes (filesystem) storage
+//! for §11.6 group-shared FILES collections. A near-clone of [`crate::files_repo`]
+//! keyed by the owning `group_id` instead of `app_id`.
+//!
+//! The security-sensitive disk mechanics — the atomic write+checksum protocol and
+//! checksum-on-read — are **not** duplicated: they come from the owner-relative
+//! free functions in [`crate::files_repo`] (`write_atomic_at` / `read_verify_at` /
+//! `final_path_at`), called with this repo's owner sub-path. Group blobs shard at
+//! `<root>/files/groups/<group_id>/<collection>/<id[0:2]>/<id>` — a `groups/` infix
+//! disjoint from the per-app `files/<app_id>/...` subtree, so the orphan sweeper
+//! ([crate::files_sweep]) covers both with one walk. Authorization, group
+//! resolution, value validation, and content-type sanitization live one layer up
+//! in `GroupFilesServiceImpl`.
+
+use std::path::{Path, PathBuf};
+
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use picloud_shared::{FileMeta, FileUpdate, FilesListPage, GroupId, NewFile};
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::files_repo::{
+    decode_cursor, encode_cursor, final_path_at, read_verify_at, write_atomic_at, FileUpdated,
+    FilesRepoError,
+};
+
+const FILES_LIST_MAX_LIMIT: u32 = 1_000;
+const FILES_LIST_DEFAULT_LIMIT: u32 = 100;
+
+#[derive(Debug, thiserror::Error)]
+pub enum GroupFilesRepoError {
+    #[error("database error: {0}")]
+    Db(#[from] sqlx::Error),
+
+    #[error("filesystem error: {0}")]
+    Io(String),
+
+    #[error("invalid collection name: {0}")]
+    InvalidCollection(String),
+
+    /// Bytes on disk no longer match the stored checksum (or are missing).
+    #[error("file content corrupted (checksum mismatch)")]
+    Corrupted,
+
+    #[error("invalid pagination cursor")]
+    InvalidCursor,
+}
+
+impl From<FilesRepoError> for GroupFilesRepoError {
+    fn from(e: FilesRepoError) -> Self {
+        match e {
+            FilesRepoError::Db(e) => Self::Db(e),
+            FilesRepoError::Io(s) => Self::Io(s),
+            FilesRepoError::InvalidCollection(c) => Self::InvalidCollection(c),
+            FilesRepoError::Corrupted => Self::Corrupted,
+            FilesRepoError::InvalidCursor => Self::InvalidCursor,
+        }
+    }
+}
+
+/// The owner-relative subdirectory of a **group**'s shared blobs under
+/// `<root>/files/`: `groups/<group_id>`. A UUID app-dir can never equal the
+/// literal `groups`, so app and group blob trees can't collide.
+pub(crate) fn group_owner_dir(group_id: GroupId) -> PathBuf {
+    Path::new("groups").join(group_id.into_inner().to_string())
+}
+
+#[async_trait]
+pub trait GroupFilesRepo: Send + Sync {
+    async fn create(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        new: NewFile,
+    ) -> Result<FileMeta, GroupFilesRepoError>;
+
+    async fn head(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        id: Uuid,
+    ) -> Result<Option<FileMeta>, GroupFilesRepoError>;
+
+    /// Reads + checksum-verifies the bytes. `Ok(None)` when no row exists;
+    /// `Err(Corrupted)` when the row exists but the bytes are missing/mismatched.
+    async fn get(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        id: Uuid,
+    ) -> Result<Option<Vec<u8>>, GroupFilesRepoError>;
+
+    /// `Ok(None)` when no row exists (the service maps that to `NotFound`).
+    async fn update(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        id: Uuid,
+        upd: FileUpdate,
+    ) -> Result<Option<FileUpdated>, GroupFilesRepoError>;
+
+    async fn delete(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        id: Uuid,
+    ) -> Result<Option<FileMeta>, GroupFilesRepoError>;
+
+    async fn list(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        cursor: Option<&str>,
+        limit: u32,
+    ) -> Result<FilesListPage, GroupFilesRepoError>;
+}
+
+/// Filesystem-bytes + Postgres-metadata repo for group-shared files.
+pub struct FsGroupFilesRepo {
+    pool: PgPool,
+    root: PathBuf,
+}
+
+impl FsGroupFilesRepo {
+    #[must_use]
+    pub fn new(pool: PgPool, root: PathBuf) -> Self {
+        Self { pool, root }
+    }
+
+    /// Belt-and-suspenders path guard (the service validates at the SDK
+    /// boundary). Mirrors `FsFilesRepo::guard_collection`.
+    fn guard_collection(collection: &str) -> Result<(), GroupFilesRepoError> {
+        if collection.is_empty()
+            || collection.contains('/')
+            || collection.contains('\\')
+            || collection.contains("..")
+            || collection.contains('\0')
+        {
+            return Err(GroupFilesRepoError::InvalidCollection(
+                collection.to_string(),
+            ));
+        }
+        Ok(())
+    }
+
+    fn write_atomic(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        id: Uuid,
+        bytes: &[u8],
+    ) -> Result<String, GroupFilesRepoError> {
+        Ok(write_atomic_at(
+            &self.root,
+            &group_owner_dir(group_id),
+            collection,
+            id,
+            bytes,
+        )?)
+    }
+}
+
+#[async_trait]
+impl GroupFilesRepo for FsGroupFilesRepo {
+    async fn create(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        new: NewFile,
+    ) -> Result<FileMeta, GroupFilesRepoError> {
+        Self::guard_collection(collection)?;
+        let id = Uuid::new_v4();
+        let size = i64::try_from(new.data.len()).unwrap_or(i64::MAX);
+        let checksum = self.write_atomic(group_id, collection, id, &new.data)?;
+
+        let row: GroupFileRow = sqlx::query_as(
+            "INSERT INTO group_files \
+                (group_id, collection, id, name, content_type, size_bytes, checksum_sha256) \
+             VALUES ($1, $2, $3, $4, $5, $6, $7) \
+             RETURNING id, collection, name, content_type, size_bytes, \
+                       checksum_sha256, created_at, updated_at",
+        )
+        .bind(group_id.into_inner())
+        .bind(collection)
+        .bind(id)
+        .bind(&new.name)
+        .bind(&new.content_type)
+        .bind(size)
+        .bind(&checksum)
+        .fetch_one(&self.pool)
+        .await?;
+        Ok(row.into_meta())
+    }
+
+    async fn head(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        id: Uuid,
+    ) -> Result<Option<FileMeta>, GroupFilesRepoError> {
+        Self::guard_collection(collection)?;
+        let row: Option<GroupFileRow> = sqlx::query_as(
+            "SELECT id, collection, name, content_type, size_bytes, \
+                    checksum_sha256, created_at, updated_at \
+             FROM group_files WHERE group_id = $1 AND collection = $2 AND id = $3",
+        )
+        .bind(group_id.into_inner())
+        .bind(collection)
+        .bind(id)
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(row.map(GroupFileRow::into_meta))
+    }
+
+    async fn get(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        id: Uuid,
+    ) -> Result<Option<Vec<u8>>, GroupFilesRepoError> {
+        Self::guard_collection(collection)?;
+        let row: Option<(String,)> = sqlx::query_as(
+            "SELECT checksum_sha256 FROM group_files \
+             WHERE group_id = $1 AND collection = $2 AND id = $3",
+        )
+        .bind(group_id.into_inner())
+        .bind(collection)
+        .bind(id)
+        .fetch_optional(&self.pool)
+        .await?;
+        let Some((stored_checksum,)) = row else {
+            return Ok(None);
+        };
+        let bytes = read_verify_at(
+            &self.root,
+            &group_owner_dir(group_id),
+            collection,
+            id,
+            &stored_checksum,
+        )?;
+        Ok(Some(bytes))
+    }
+
+    async fn update(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        id: Uuid,
+        upd: FileUpdate,
+    ) -> Result<Option<FileUpdated>, GroupFilesRepoError> {
+        Self::guard_collection(collection)?;
+        let Some(prev) = self.head(group_id, collection, id).await? else {
+            return Ok(None);
+        };
+        let size = i64::try_from(upd.data.len()).unwrap_or(i64::MAX);
+        let checksum = self.write_atomic(group_id, collection, id, &upd.data)?;
+
+        let row: GroupFileRow = sqlx::query_as(
+            "UPDATE group_files SET \
+                name = COALESCE($4, name), \
+                content_type = COALESCE($5, content_type), \
+                size_bytes = $6, \
+                checksum_sha256 = $7, \
+                updated_at = NOW() \
+             WHERE group_id = $1 AND collection = $2 AND id = $3 \
+             RETURNING id, collection, name, content_type, size_bytes, \
+                       checksum_sha256, created_at, updated_at",
+        )
+        .bind(group_id.into_inner())
+        .bind(collection)
+        .bind(id)
+        .bind(upd.name.as_deref())
+        .bind(upd.content_type.as_deref())
+        .bind(size)
+        .bind(&checksum)
+        .fetch_one(&self.pool)
+        .await?;
+        Ok(Some(FileUpdated {
+            new: row.into_meta(),
+            prev,
+        }))
+    }
+
+    async fn delete(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        id: Uuid,
+    ) -> Result<Option<FileMeta>, GroupFilesRepoError> {
+        Self::guard_collection(collection)?;
+        let mut tx = self.pool.begin().await?;
+        let row: Option<GroupFileRow> = sqlx::query_as(
+            "SELECT id, collection, name, content_type, size_bytes, \
+                    checksum_sha256, created_at, updated_at \
+             FROM group_files WHERE group_id = $1 AND collection = $2 AND id = $3 \
+             FOR UPDATE",
+        )
+        .bind(group_id.into_inner())
+        .bind(collection)
+        .bind(id)
+        .fetch_optional(&mut *tx)
+        .await?;
+
+        let Some(row) = row else {
+            tx.rollback().await?;
+            return Ok(None);
+        };
+
+        sqlx::query("DELETE FROM group_files WHERE group_id = $1 AND collection = $2 AND id = $3")
+            .bind(group_id.into_inner())
+            .bind(collection)
+            .bind(id)
+            .execute(&mut *tx)
+            .await?;
+        tx.commit().await?;
+
+        // Row gone; unlink the bytes. A failure here leaves an orphan file
+        // (reclaimed by the sweep) — not fatal.
+        let path = final_path_at(&self.root, &group_owner_dir(group_id), collection, id);
+        if let Err(e) = std::fs::remove_file(&path) {
+            if e.kind() != std::io::ErrorKind::NotFound {
+                tracing::warn!(path = %path.display(), error = %e, "group files: unlink after delete failed (orphan)");
+            }
+        }
+        Ok(Some(row.into_meta()))
+    }
+
+    async fn list(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        cursor: Option<&str>,
+        limit: u32,
+    ) -> Result<FilesListPage, GroupFilesRepoError> {
+        Self::guard_collection(collection)?;
+        let limit = if limit == 0 {
+            FILES_LIST_DEFAULT_LIMIT
+        } else {
+            limit.min(FILES_LIST_MAX_LIMIT)
+        };
+        let last_id = match cursor {
+            Some(c) => Some(decode_cursor(c)?),
+            None => None,
+        };
+        let take = i64::from(limit) + 1;
+        let rows: Vec<GroupFileRow> = sqlx::query_as(
+            "SELECT id, collection, name, content_type, size_bytes, \
+                    checksum_sha256, created_at, updated_at \
+             FROM group_files \
+             WHERE group_id = $1 AND collection = $2 \
+               AND ($3::uuid IS NULL OR id > $3) \
+             ORDER BY id ASC \
+             LIMIT $4",
+        )
+        .bind(group_id.into_inner())
+        .bind(collection)
+        .bind(last_id)
+        .bind(take)
+        .fetch_all(&self.pool)
+        .await?;
+
+        let mut files: Vec<FileMeta> = rows.into_iter().map(GroupFileRow::into_meta).collect();
+        let next_cursor = if files.len() > limit as usize {
+            files.truncate(limit as usize);
+            files.last().map(|m| encode_cursor(m.id))
+        } else {
+            None
+        };
+        Ok(FilesListPage { files, next_cursor })
+    }
+}
+
+#[derive(sqlx::FromRow)]
+struct GroupFileRow {
+    id: Uuid,
+    collection: String,
+    name: String,
+    content_type: String,
+    size_bytes: i64,
+    checksum_sha256: String,
+    created_at: DateTime<Utc>,
+    updated_at: DateTime<Utc>,
+}
+
+impl GroupFileRow {
+    fn into_meta(self) -> FileMeta {
+        FileMeta {
+            id: self.id,
+            collection: self.collection,
+            name: self.name,
+            content_type: self.content_type,
+            size: u64::try_from(self.size_bytes).unwrap_or(0),
+            checksum: self.checksum_sha256,
+            created_at: self.created_at,
+            updated_at: self.updated_at,
+        }
+    }
+}

crates/manager-core/src/lib.rs

@@ -53,6 +53,7 @@ pub mod gc;
 pub mod group_collection_repo;
 pub mod group_docs_repo;
 pub mod group_docs_service;
+pub mod group_files_repo;
 pub mod group_kv_repo;
 pub mod group_kv_service;
 pub mod group_members_repo;
@@ -184,6 +185,7 @@ pub use gc::{spawn_abandoned_gc, spawn_app_user_token_gc, spawn_dead_letter_gc};
 pub use group_collection_repo::{GroupCollectionResolver, PostgresGroupCollectionResolver};
 pub use group_docs_repo::{GroupDocsRepo, GroupDocsRepoError, PostgresGroupDocsRepo};
 pub use group_docs_service::GroupDocsServiceImpl;
+pub use group_files_repo::{FsGroupFilesRepo, GroupFilesRepo, GroupFilesRepoError};
 pub use group_kv_repo::{GroupKvRepo, GroupKvRepoError, PostgresGroupKvRepo};
 pub use group_kv_service::GroupKvServiceImpl;
 pub use group_members_repo::{

crates/manager-core/tests/expected_schema.txt

@@ -239,6 +239,17 @@ table: group_docs
   created_at: timestamp with time zone NOT NULL default=now()
   updated_at: timestamp with time zone NOT NULL default=now()
 
+table: group_files
+  group_id: uuid NOT NULL
+  collection: text NOT NULL
+  id: uuid NOT NULL
+  name: text NOT NULL
+  content_type: text NOT NULL
+  size_bytes: bigint NOT NULL
+  checksum_sha256: text NOT NULL
+  created_at: timestamp with time zone NOT NULL default=now()
+  updated_at: timestamp with time zone NOT NULL default=now()
+
 table: group_kv_entries
   group_id: uuid NOT NULL
   collection: text NOT NULL
@@ -522,6 +533,10 @@ indexes on group_docs:
   idx_group_docs_data_gin: public.group_docs USING gin (data jsonb_path_ops)
   idx_group_docs_group_collection: public.group_docs USING btree (group_id, collection)
 
+indexes on group_files:
+  group_files_pkey: public.group_files USING btree (group_id, collection, id)
+  idx_group_files_group_collection: public.group_files USING btree (group_id, collection)
+
 indexes on group_kv_entries:
   group_kv_entries_pkey: public.group_kv_entries USING btree (group_id, collection, key)
   idx_group_kv_entries_group_collection: public.group_kv_entries USING btree (group_id, collection)
@@ -727,7 +742,7 @@ constraints on files_trigger_details:
   [PRIMARY KEY] files_trigger_details_pkey: PRIMARY KEY (trigger_id)
 
 constraints on group_collections:
-  [CHECK] group_collections_kind_check: CHECK ((kind = ANY (ARRAY['kv'::text, 'docs'::text])))
+  [CHECK] group_collections_kind_check: CHECK ((kind = ANY (ARRAY['kv'::text, 'docs'::text, 'files'::text])))
   [CHECK] group_collections_owner_exactly_one: CHECK (((group_id IS NULL) <> (app_id IS NULL)))
   [FOREIGN KEY] group_collections_app_id_fkey: FOREIGN KEY (app_id) REFERENCES apps(id) ON DELETE CASCADE
   [FOREIGN KEY] group_collections_group_id_fkey: FOREIGN KEY (group_id) REFERENCES groups(id) ON DELETE CASCADE
@@ -737,6 +752,10 @@ constraints on group_docs:
   [FOREIGN KEY] group_docs_group_id_fkey: FOREIGN KEY (group_id) REFERENCES groups(id) ON DELETE CASCADE
   [PRIMARY KEY] group_docs_pkey: PRIMARY KEY (group_id, collection, id)
 
+constraints on group_files:
+  [FOREIGN KEY] group_files_group_id_fkey: FOREIGN KEY (group_id) REFERENCES groups(id) ON DELETE CASCADE
+  [PRIMARY KEY] group_files_pkey: PRIMARY KEY (group_id, collection, id)
+
 constraints on group_kv_entries:
   [FOREIGN KEY] group_kv_entries_group_id_fkey: FOREIGN KEY (group_id) REFERENCES groups(id) ON DELETE CASCADE
   [PRIMARY KEY] group_kv_entries_pkey: PRIMARY KEY (group_id, collection, key)
@@ -881,3 +900,4 @@ constraints on vars:
   0052: group collections
   0053: group kv entries
   0054: group docs
+  0055: group files