halkyon/halkyon_learn_server/test/integration/asset/asset_test.dart

import 'dart:convert';
import 'dart:typed_data';

import 'package:crypto/crypto.dart';
import 'package:halkyon_learn_server/src/asset/asset_endpoint.dart';
import 'package:halkyon_learn_server/src/asset/file_limits.dart';
import 'package:halkyon_learn_server/src/asset/file_pruning.dart';
import 'package:halkyon_learn_server/src/auth/scopes.dart';
import 'package:halkyon_learn_server/src/generated/protocol.dart';
import 'package:serverpod/protocol.dart' show CloudStorageEntry;
import 'package:serverpod/serverpod.dart';
import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
import 'package:test/test.dart';

import '../test_tools/serverpod_test_tools.dart';

void main() {
  withServerpod(
    'Files',
    (sessionBuilder, endpoints) {
      late Session session;
      late UuidValue alice;
      late UuidValue bob;
      final image = utf8.encode('an image');
      final imageHash = sha256.convert(image).toString();

      /// The spaces a test made, which it deletes again with their files.
      final spaces = <UuidValue>{};

      setUp(() {
        // The test pod is built without the generated Serverpod class, which
        // sets up sync for the running server.
        Serverpod.instance.initializeOfflineSync(syncTables: syncTables);
        session = sessionBuilder.build();
        alice = const Uuid().v7obj();
        bob = const Uuid().v7obj();
        spaces.addAll([alice, bob]);
      });

      tearDown(() async {
        fileLimits = null;
        await FileAllowance.db.deleteWhere(
          session,
          where: (t) => t.authUserId.inSet(spaces),
        );
        for (final space in spaces) {
          await session.db.unsafeExecute(
            'DELETE FROM "serverpod_cloud_storage" WHERE "path" LIKE @prefix',
            parameters: QueryParameters.named({
              'prefix': '${assetPath(space, '')}%',
            }),
          );
          await session.db.unsafeExecute(
            'DELETE FROM "offline_sync_spaces" WHERE "uuidSpaceId" = @space',
            parameters: QueryParameters.named({'space': space.uuid}),
          );
        }
        spaces.clear();
      });

      TestSessionBuilder signedIn(
        UuidValue person, {
        Set<Scope> scopes = const {},
      }) => sessionBuilder.copyWith(
        authentication: AuthenticationOverride.authenticationInfo(
          person.uuid,
          scopes,
        ),
      );

      /// Adds [bytes] as [person], as the app does: asks how to upload them,
      /// uploads them unless the server has them, and finishes the upload.
      Future<bool> add(
        UuidValue person,
        List<int> bytes, {
        String? as,
        Set<Scope> scopes = const {},
      }) async {
        final hash = as ?? sha256.convert(bytes).toString();
        final description = await endpoints.asset.startUpload(
          signedIn(person, scopes: scopes),
          hash,
          bytes.length,
        );
        if (description != null) {
          await _upload(session, description, bytes);
        }
        return await endpoints.asset.finishUpload(
          signedIn(person, scopes: scopes),
          hash,
        );
      }

      test('refuses callers who are not signed in', () async {
        final nobody = sessionBuilder.copyWith(
          authentication: AuthenticationOverride.unauthenticated(),
        );
        final unauthenticated = throwsA(
          isA<ServerpodUnauthenticatedException>(),
        );

        await expectLater(
          endpoints.asset.startUpload(nobody, imageHash, image.length),
          unauthenticated,
        );
        await expectLater(
          endpoints.asset.finishUpload(nobody, imageHash),
          unauthenticated,
        );
        await expectLater(
          endpoints.asset.download(nobody, const Uuid().v7obj()),
          unauthenticated,
        );
      });

      test('stores a file once and serves it by its asset row', () async {
        final asset = await _assetIn(session, alice, imageHash);

        expect(await add(alice, image), isTrue);

        expect(
          await endpoints.asset.startUpload(
            signedIn(alice),
            imageHash,
            image.length,
          ),
          isNull,
        );
        expect(
          Uint8List.sublistView(
            await endpoints.asset.download(signedIn(alice), asset),
          ),
          image,
        );
      });

      test('keeps no file whose content has another hash', () async {
        final asset = await _assetIn(session, alice, imageHash);

        expect(
          await add(alice, utf8.encode('not that image'), as: imageHash),
          isFalse,
        );

        await expectLater(
          endpoints.asset.download(signedIn(alice), asset),
          _notFound,
        );
        expect(
          await endpoints.asset.startUpload(
            signedIn(alice),
            imageHash,
            image.length,
          ),
          isNotNull,
        );
      });

      test('keeps a file over 20 MB only when it is a PDF, which may take '
          'up to 50 MB', () async {
        const size = AssetEndpoint.maxOtherBytes + 1;
        final pdf = Uint8List(size)
          ..setAll(0, utf8.encode('%PDF-1.7'))
          ..fillRange(8, size, 32);
        final other = Uint8List(size)..fillRange(0, size, 32);
        final otherHash = sha256.convert(other).toString();

        expect(await add(alice, pdf), isTrue);
        expect(await add(alice, other), isFalse);

        // What isn't a PDF is gone: the server asks for it anew.
        expect(
          await endpoints.asset.startUpload(signedIn(alice), otherHash, size),
          isNotNull,
        );
        await expectLater(
          endpoints.asset.startUpload(
            signedIn(alice),
            otherHash,
            AssetEndpoint.maxBytes + 1,
          ),
          throwsArgumentError,
        );
      });

      test('finishes no upload that never came', () async {
        expect(
          await endpoints.asset.finishUpload(signedIn(alice), imageHash),
          isFalse,
        );
      });

      test('refuses what is no SHA-256, and sizes out of range', () async {
        for (final hash in [
          '',
          'abc',
          imageHash.toUpperCase(),
          '$imageHash 0',
        ]) {
          await expectLater(
            endpoints.asset.startUpload(signedIn(alice), hash, 1),
            throwsArgumentError,
          );
          await expectLater(
            endpoints.asset.finishUpload(signedIn(alice), hash),
            throwsArgumentError,
          );
        }
        for (final size in [0, -1, AssetEndpoint.maxBytes + 1]) {
          await expectLater(
            endpoints.asset.startUpload(signedIn(alice), imageHash, size),
            throwsArgumentError,
          );
        }
      });

      test(
        "serves nobody else's file, exactly like a missing one",
        () async {
          final asset = await _assetIn(session, alice, imageHash);
          await add(alice, image);
          final notUploaded = await _assetIn(
            session,
            alice,
            sha256.convert([1]).toString(),
          );

          await expectLater(
            endpoints.asset.download(signedIn(bob), asset),
            _notFound,
          );
          await expectLater(
            endpoints.asset.download(signedIn(alice), notUploaded),
            _notFound,
          );
          await expectLater(
            endpoints.asset.download(signedIn(alice), const Uuid().v7obj()),
            _notFound,
          );
        },
      );

      test(
        "keeps every person's files apart: the same file from someone else "
        'neither shows nor replaces yours',
        () async {
          final asset = await _assetIn(session, alice, imageHash);
          await add(alice, image);

          // Bob is asked for the file as if nobody had it.
          expect(
            await endpoints.asset.startUpload(
              signedIn(bob),
              imageHash,
              image.length,
            ),
            isNotNull,
          );
          expect(
            await add(bob, utf8.encode('not that image'), as: imageHash),
            isFalse,
          );

          expect(
            Uint8List.sublistView(
              await endpoints.asset.download(signedIn(alice), asset),
            ),
            image,
          );
        },
      );

      test(
        "serves a shared space's files to its members, and only them",
        () async {
          final carol = const Uuid().v7obj();
          final spaces = OfflineSyncSpaces(session);
          final shared = await spaces.createFor(carol);
          await spaces.grant(
            user: alice,
            space: shared,
            role: OfflineSyncSpaceRole.readOnly,
          );
          final asset = await _assetIn(
            session,
            carol,
            imageHash,
            space: shared,
          );
          await session.storage.storeFile(
            storageId: assetStorage,
            path: assetPath(shared, imageHash),
            byteData: ByteData.sublistView(image),
          );

          expect(
            Uint8List.sublistView(
              await endpoints.asset.download(signedIn(alice), asset),
            ),
            image,
          );
          await expectLater(
            endpoints.asset.download(signedIn(bob), asset),
            _notFound,
          );
        },
      );

      test('serves no deleted asset', () async {
        final asset = await _assetIn(session, alice, imageHash);
        await add(alice, image);

        await _deleteAsset(session, alice, asset);

        await expectLater(
          endpoints.asset.download(signedIn(alice), asset),
          _notFound,
        );
      });

      group('room', () {
        setUp(() {
          fileLimits = const FileLimits(
            anonymousBytes: 10,
            identifiedBytes: 20,
            instanceBytes: 1000,
          );
        });

        Matcher refused({required bool instanceFull, required int used}) =>
            throwsA(
              isA<FileLimitExceededException>()
                  .having((e) => e.instanceFull, 'instanceFull', instanceFull)
                  .having((e) => e.usedBytes, 'usedBytes', used),
            );

        test('an anonymous person keeps no more than their room', () async {
          expect(await add(alice, List.filled(8, 1)), isTrue);

          await expectLater(
            endpoints.asset.startUpload(
              signedIn(alice),
              sha256.convert([2, 2, 2]).toString(),
              3,
            ),
            refused(instanceFull: false, used: 8),
          );
        });

        test('an identified person keeps more', () async {
          expect(
            await add(alice, List.filled(15, 1), scopes: {identified}),
            isTrue,
          );
          await expectLater(
            add(bob, List.filled(15, 1)),
            refused(instanceFull: false, used: 0),
          );
        });

        test('an allowance gives one person more, and nobody else', () async {
          await FileAllowance.db.insertRow(
            session,
            FileAllowance(authUserId: alice, maxBytes: 40),
          );

          expect(await add(alice, List.filled(30, 1)), isTrue);
          await expectLater(
            add(bob, List.filled(30, 2)),
            refused(instanceFull: false, used: 0),
          );
        });

        test('the instance keeps no more than its room', () async {
          // Whatever other tests left behind counts too: the room is set
          // from what the instance holds now.
          fileLimits = FileLimits(
            anonymousBytes: 10,
            identifiedBytes: 20,
            instanceBytes: await _instanceBytes(session) + 12,
          );
          expect(await add(alice, List.filled(8, 1)), isTrue);

          await expectLater(
            add(bob, List.filled(5, 2)),
            refused(instanceFull: true, used: 0),
          );
        });

        test('uploads started side by side are checked again as they '
            'finish', () async {
          final first = List.filled(6, 1);
          final second = List.filled(6, 2);
          String hash(List<int> bytes) => sha256.convert(bytes).toString();
          // Both started before either arrives: each fits on its own.
          final descriptions = [
            for (final bytes in [first, second])
              (await endpoints.asset.startUpload(
                signedIn(alice),
                hash(bytes),
                bytes.length,
              ))!,
          ];
          await _upload(session, descriptions[0], first);
          await _upload(session, descriptions[1], second);

          expect(
            await endpoints.asset.finishUpload(signedIn(alice), hash(first)),
            isTrue,
          );
          await expectLater(
            endpoints.asset.finishUpload(signedIn(alice), hash(second)),
            refused(instanceFull: false, used: 12),
          );
          expect(
            await session.storage.fileExists(
              storageId: assetStorage,
              path: assetPath(alice, hash(second)),
            ),
            isFalse,
          );
        });

        test('uploads never finished take room from the next', () async {
          final unfinished = List.filled(6, 1);
          await _upload(
            session,
            (await endpoints.asset.startUpload(
              signedIn(alice),
              sha256.convert(unfinished).toString(),
              unfinished.length,
            ))!,
            unfinished,
          );

          await expectLater(
            endpoints.asset.startUpload(
              signedIn(alice),
              sha256.convert(List.filled(6, 2)).toString(),
              6,
            ),
            refused(instanceFull: false, used: 6),
          );
        });

        test('tells how much room the files take, and may take', () async {
          await add(alice, List.filled(4, 1));

          final usage = await endpoints.asset.usage(signedIn(alice));
          expect((usage.usedBytes, usage.limitBytes), (4, 10));

          fileLimits = FileLimits(
            anonymousBytes: 10,
            identifiedBytes: 20,
            instanceBytes: await _instanceBytes(session) + 2,
          );
          final tight = await endpoints.asset.usage(signedIn(alice));
          expect((tight.usedBytes, tight.limitBytes), (4, 6));
        });

        test('refuses callers who are not signed in', () async {
          await expectLater(
            endpoints.asset.usage(
              sessionBuilder.copyWith(
                authentication: AuthenticationOverride.unauthenticated(),
              ),
            ),
            throwsA(isA<ServerpodUnauthenticatedException>()),
          );
        });
      });

      group('pruning', () {
        final now = DateTime.now().toUtc();
        DateTime daysAgo(int days) => now.subtract(Duration(days: days));

        Future<bool> stored(UuidValue space, String hash) async =>
            await CloudStorageEntry.db.count(
              session,
              where: (t) =>
                  t.storageId.equals(assetStorage) &
                  t.path.equals(assetPath(space, hash)),
            ) >
            0;

        test(
          'deletes the files no asset refers to any more, a week after they '
          'were last claimed',
          () async {
            final kept = utf8.encode('kept');
            final deleted = utf8.encode('deleted');
            final recent = utf8.encode('recently uploaded');
            final unfinished = utf8.encode('never finished');
            final bobs = utf8.encode("bob's");
            String hash(List<int> bytes) => sha256.convert(bytes).toString();
            // Whatever an earlier, aborted run left behind goes first.
            await pruneUnreferencedFiles(session, now: now);

            await _assetIn(session, alice, hash(kept));
            await add(alice, kept);
            final gone = await _assetIn(session, alice, hash(deleted));
            await add(alice, deleted);
            await _deleteAsset(session, alice, gone);
            await add(alice, recent);
            await _upload(
              session,
              (await endpoints.asset.startUpload(
                signedIn(alice),
                hash(unfinished),
                unfinished.length,
              ))!,
              unfinished,
            );
            await _assetIn(session, alice, hash(unfinished));
            await add(bob, bobs);
            for (final (space, bytes) in [
              (alice, kept),
              (alice, deleted),
              (alice, unfinished),
              (bob, bobs),
            ]) {
              await _claimed(session, space, hash(bytes), daysAgo(8));
            }
            await _claimed(session, alice, hash(recent), daysAgo(6));

            expect(await pruneUnreferencedFiles(session, now: now), 3);

            expect(await stored(alice, hash(kept)), isTrue);
            expect(await stored(alice, hash(recent)), isTrue);
            expect(await stored(alice, hash(deleted)), isFalse);
            expect(await stored(alice, hash(unfinished)), isFalse);
            expect(await stored(bob, hash(bobs)), isFalse);
          },
        );

        test('keeps a file asked for again', () async {
          await add(alice, image);
          await _claimed(session, alice, imageHash, daysAgo(30));

          expect(
            await endpoints.asset.startUpload(
              signedIn(alice),
              imageHash,
              image.length,
            ),
            isNull,
          );

          expect(await pruneUnreferencedFiles(session, now: now), 0);
          expect(await stored(alice, imageHash), isTrue);
        });
      });
    },
    // The sync engine's database, as the running server has it: deleted rows
    // are tombstones it hides. Writing through it takes transactions of its
    // own, outside the one a test would be rolled back in, so the tests
    // delete what they made instead.
    databaseInterceptor: (session, inner) {
      final database = offlineSyncDatabaseInterceptor(session, inner);
      if (database is OfflineSyncDatabase) {
        _syncDatabases[session] = database;
      }
      return database;
    },
    rollbackDatabase: RollbackDatabase.disabled,
  );
}

/// The sync engine's database of each session, as the test pod's own
/// database wraps it.
final _syncDatabases = Expando<OfflineSyncDatabase>();

final _notFound = throwsA(
  isA<NotFoundException>().having((e) => e.what, 'what', 'file'),
);

/// An asset row for the file with [hash], written as [person] would through
/// the sync engine, in their personal space or in [space]; returns its id.
Future<UuidValue> _assetIn(
  Session session,
  UuidValue person,
  String hash, {
  UuidValue? space,
}) => _syncDatabases[session]!.transactionForUser(
  person,
  spaceId: space,
  (transaction) async => (await Asset.db.insertRow(
    session,
    Asset(
      sha256: hash,
      mimeType: 'image/png',
      byteSize: 1,
      payload: ByteData(0),
    ),
    transaction: transaction,
  )).id!,
);

/// Deletes the asset row [asset] as [person] would through the sync engine.
Future<void> _deleteAsset(
  Session session,
  UuidValue person,
  UuidValue asset,
) => _syncDatabases[session]!.transactionForUser(
  person,
  (transaction) => Asset.db.deleteWhere(
    session,
    where: (t) => t.id.equals(asset),
    transaction: transaction,
  ),
);

/// Uploads [bytes] as the upload [description] says, standing in for the
/// app's HTTP upload: the server's upload route stores the file unverified at
/// the path the description names, if its key is the one issued.
Future<void> _upload(
  Session session,
  String description,
  List<int> bytes,
) async {
  final url = Uri.parse(
    (jsonDecode(description) as Map<String, Object?>)['url']! as String,
  );
  final storageId = url.queryParameters['storage']!;
  final path = url.queryParameters['path']!;
  final issued = await session.db.unsafeQuery(
    'SELECT 1 FROM "serverpod_cloud_storage_direct_upload" '
    'WHERE "storageId" = @storage AND "path" = @path AND "authKey" = @key',
    parameters: QueryParameters.named({
      'storage': storageId,
      'path': path,
      'key': url.queryParameters['key'],
    }),
  );
  expect(issued, isNotEmpty, reason: 'the upload was authorized');
  final storage =
      Serverpod.instance.storage[storageId]! as DatabaseCloudStorage;
  await storage.storeUnverifiedFile(
    session: session,
    path: path,
    byteData: ByteData.sublistView(Uint8List.fromList(bytes)),
  );
}

/// What the stored files of everyone take, in bytes.
Future<int> _instanceBytes(Session session) async =>
    (await session.db.unsafeQuery(
          'SELECT COALESCE(SUM(octet_length("byteData")), 0)::bigint '
          'FROM "serverpod_cloud_storage" WHERE "path" LIKE \'spaces/%\'',
        )).first.first!
        as int;

/// Makes the stored file with [hash] of [space] last claimed [at].
Future<void> _claimed(
  Session session,
  UuidValue space,
  String hash,
  DateTime at,
) => CloudStorageEntry.db.updateWhere(
  session,
  columnValues: (t) => [t.addedTime(at)],
  where: (t) =>
      t.storageId.equals(assetStorage) & t.path.equals(assetPath(space, hash)),
);