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_pruning.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 {
        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) => sessionBuilder.copyWith(
        authentication: AuthenticationOverride.authenticationInfo(
          person.uuid,
          {},
        ),
      );

      /// 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}) async {
        final hash = as ?? sha256.convert(bytes).toString();
        final description = await endpoints.asset.startUpload(
          signedIn(person),
          hash,
          bytes.length,
        );
        if (description != null) {
          await _upload(session, description, bytes);
        }
        return await endpoints.asset.finishUpload(signedIn(person), 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('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('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)),
  );
}

/// 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)),
);