amduat-api/tests/test_amduatd_space_mounts_sync.c

#ifndef _POSIX_C_SOURCE
#define _POSIX_C_SOURCE 200809L
#endif

#include "amduatd_space_mounts_sync.h"

#include "amduatd_fed_cursor.h"
#include "amduatd_space.h"
#include "amduatd_space_manifest.h"
#include "amduatd_store.h"

#include "amduat/asl/asl_pointer_fs.h"
#include "amduat/asl/asl_store_fs_meta.h"
#include "amduat/asl/ref_text.h"
#include "amduat/hash/asl1.h"

#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

typedef struct {
  size_t calls;
} amduatd_test_sync_transport_t;

static int failures = 0;

static void expect(bool cond, const char *msg) {
  if (!cond) {
    fprintf(stderr, "FAIL: %s\n", msg);
    failures++;
  }
}

static char *amduatd_test_make_temp_dir(void) {
  char tmpl[] = "/tmp/amduatd-space-mounts-sync-XXXXXX";
  char *dir = mkdtemp(tmpl);
  size_t len;
  char *copy;
  if (dir == NULL) {
    perror("mkdtemp");
    return NULL;
  }
  len = strlen(dir);
  copy = (char *)malloc(len + 1u);
  if (copy == NULL) {
    fprintf(stderr, "failed to allocate temp dir copy\n");
    return NULL;
  }
  memcpy(copy, dir, len + 1u);
  return copy;
}

static bool amduatd_make_test_ref(uint8_t fill, amduat_reference_t *out_ref) {
  uint8_t digest_bytes[32];
  amduat_octets_t digest;
  if (out_ref == NULL) {
    return false;
  }
  memset(digest_bytes, fill, sizeof(digest_bytes));
  if (!amduat_octets_clone(amduat_octets(digest_bytes, sizeof(digest_bytes)),
                           &digest)) {
    return false;
  }
  *out_ref = amduat_reference(AMDUAT_HASH_ASL1_ID_SHA256, digest);
  return true;
}

static bool amduatd_test_get_records(void *ctx,
                                     uint32_t domain_id,
                                     uint64_t from_logseq,
                                     uint64_t limit,
                                     int *out_status,
                                     amduat_fed_record_t **out_records,
                                     size_t *out_len,
                                     char **out_body) {
  amduatd_test_sync_transport_t *t = (amduatd_test_sync_transport_t *)ctx;
  (void)domain_id;
  (void)from_logseq;
  (void)limit;
  if (out_status == NULL || out_records == NULL || out_len == NULL) {
    return false;
  }
  *out_status = 200;
  *out_records = NULL;
  *out_len = 0u;
  if (out_body != NULL) {
    *out_body = NULL;
  }
  if (t != NULL) {
    t->calls++;
  }
  return true;
}

static void amduatd_test_free_records(void *ctx,
                                      amduat_fed_record_t *records,
                                      size_t len) {
  (void)ctx;
  (void)records;
  (void)len;
}

static bool amduatd_test_get_artifact(void *ctx,
                                      amduat_reference_t ref,
                                      int *out_status,
                                      amduat_octets_t *out_bytes,
                                      char **out_body) {
  (void)ctx;
  (void)ref;
  if (out_status == NULL || out_bytes == NULL) {
    return false;
  }
  *out_status = 404;
  *out_bytes = amduat_octets(NULL, 0u);
  if (out_body != NULL) {
    *out_body = NULL;
  }
  return true;
}

static bool amduatd_test_seed_manifest(amduat_asl_store_t *store,
                                       amduat_asl_pointer_store_t *pointer_store,
                                       const amduatd_space_t *space,
                                       const char *json) {
  amduat_reference_t ref;
  amduatd_space_manifest_t manifest;
  amduat_octets_t payload = amduat_octets(json, strlen(json));
  amduatd_space_manifest_status_t status;

  memset(&ref, 0, sizeof(ref));
  memset(&manifest, 0, sizeof(manifest));
  status = amduatd_space_manifest_put(store,
                                      pointer_store,
                                      space,
                                      payload,
                                      NULL,
                                      &ref,
                                      &manifest);
  if (status != AMDUATD_SPACE_MANIFEST_OK) {
    amduatd_space_manifest_free(&manifest);
    amduat_reference_free(&ref);
    return false;
  }
  amduatd_space_manifest_free(&manifest);
  amduat_reference_free(&ref);
  return true;
}

static bool amduatd_test_seed_cursor(amduat_asl_store_t *store,
                                     amduat_asl_pointer_store_t *pointer_store,
                                     const amduatd_space_t *space,
                                     const char *peer_key,
                                     const char *remote_space_id,
                                     uint64_t logseq,
                                     uint8_t fill) {
  amduatd_fed_cursor_record_t cursor;
  amduat_reference_t ref;
  amduatd_fed_cursor_status_t status;
  const char *space_id = NULL;
  bool ok = false;

  amduatd_fed_cursor_record_init(&cursor);
  cursor.peer_key = strdup(peer_key);
  if (space != NULL && space->enabled && space->space_id.data != NULL) {
    space_id = (const char *)space->space_id.data;
  }
  cursor.space_id = space_id != NULL ? strdup(space_id) : NULL;
  if (cursor.peer_key == NULL || cursor.space_id == NULL) {
    amduatd_fed_cursor_record_free(&cursor);
    return false;
  }
  cursor.has_logseq = true;
  cursor.last_logseq = logseq;
  if (!amduatd_make_test_ref(fill, &ref)) {
    amduatd_fed_cursor_record_free(&cursor);
    return false;
  }
  cursor.has_record_ref = true;
  cursor.last_record_ref = ref;

  status = amduatd_fed_cursor_cas_set_remote(store,
                                             pointer_store,
                                             space,
                                             peer_key,
                                             remote_space_id,
                                             NULL,
                                             &cursor,
                                             NULL);
  ok = (status == AMDUATD_FED_CURSOR_OK);
  amduatd_fed_cursor_record_free(&cursor);
  return ok;
}

static int amduatd_test_sync_missing_manifest(void) {
  char *root = amduatd_test_make_temp_dir();
  amduat_asl_store_fs_config_t cfg;
  amduatd_store_ctx_t store_ctx;
  amduat_asl_store_t store;
  amduat_asl_pointer_store_t pointer_store;
  amduatd_space_t space;
  amduatd_fed_cfg_t fed_cfg;
  amduatd_test_sync_transport_t transport_state;
  amduatd_fed_pull_transport_t transport;
  amduatd_space_mounts_sync_report_t report;
  amduatd_space_mounts_sync_status_t status;

  if (root == NULL) {
    return 1;
  }
  memset(&cfg, 0, sizeof(cfg));
  if (!amduat_asl_store_fs_init_root(root, NULL, &cfg)) {
    fprintf(stderr, "failed to init store root\n");
    free(root);
    return 1;
  }
  memset(&store_ctx, 0, sizeof(store_ctx));
  memset(&store, 0, sizeof(store));
  if (!amduatd_store_init(&store,
                          &cfg,
                          &store_ctx,
                          root,
                          AMDUATD_STORE_BACKEND_INDEX)) {
    fprintf(stderr, "failed to init store\n");
    free(root);
    return 1;
  }
  if (!amduat_asl_pointer_store_init(&pointer_store, root)) {
    fprintf(stderr, "failed to init pointer store\n");
    free(root);
    return 1;
  }
  if (!amduatd_space_init(&space, "alpha", false)) {
    fprintf(stderr, "failed to init space\n");
    free(root);
    return 1;
  }
  amduatd_fed_cfg_init(&fed_cfg);
  fed_cfg.enabled = true;

  memset(&transport_state, 0, sizeof(transport_state));
  memset(&transport, 0, sizeof(transport));
  transport.ctx = &transport_state;
  transport.get_records = amduatd_test_get_records;
  transport.free_records = amduatd_test_free_records;
  transport.get_artifact = amduatd_test_get_artifact;

  status = amduatd_space_mounts_sync_until(&store,
                                           &pointer_store,
                                           &space,
                                           &fed_cfg,
                                           &transport,
                                           128u,
                                           10u,
                                           32u,
                                           &report);
  expect(status == AMDUATD_SPACE_MOUNTS_SYNC_ERR_NOT_FOUND,
         "missing manifest");
  free(root);
  return failures == 0 ? 0 : 1;
}

static int amduatd_test_sync_no_track_mounts(void) {
  char *root = amduatd_test_make_temp_dir();
  amduat_asl_store_fs_config_t cfg;
  amduatd_store_ctx_t store_ctx;
  amduat_asl_store_t store;
  amduat_asl_pointer_store_t pointer_store;
  amduatd_space_t space;
  amduatd_fed_cfg_t fed_cfg;
  amduatd_test_sync_transport_t transport_state;
  amduatd_fed_pull_transport_t transport;
  amduat_reference_t pinned_ref;
  char *pinned_hex = NULL;
  char json[512];
  int n;
  amduatd_space_mounts_sync_report_t report;
  amduatd_space_mounts_sync_status_t status;

  if (root == NULL) {
    return 1;
  }
  memset(&cfg, 0, sizeof(cfg));
  if (!amduat_asl_store_fs_init_root(root, NULL, &cfg)) {
    fprintf(stderr, "failed to init store root\n");
    free(root);
    return 1;
  }
  memset(&store_ctx, 0, sizeof(store_ctx));
  memset(&store, 0, sizeof(store));
  if (!amduatd_store_init(&store,
                          &cfg,
                          &store_ctx,
                          root,
                          AMDUATD_STORE_BACKEND_INDEX)) {
    fprintf(stderr, "failed to init store\n");
    free(root);
    return 1;
  }
  if (!amduat_asl_pointer_store_init(&pointer_store, root)) {
    fprintf(stderr, "failed to init pointer store\n");
    free(root);
    return 1;
  }
  if (!amduatd_space_init(&space, "alpha", false)) {
    fprintf(stderr, "failed to init space\n");
    free(root);
    return 1;
  }
  amduatd_fed_cfg_init(&fed_cfg);
  fed_cfg.enabled = true;

  if (!amduatd_make_test_ref(0x11, &pinned_ref) ||
      !amduat_asl_ref_encode_hex(pinned_ref, &pinned_hex)) {
    fprintf(stderr, "failed to build pinned ref\n");
    amduat_reference_free(&pinned_ref);
    free(root);
    return 1;
  }
  n = snprintf(json, sizeof(json),
               "{\"version\":1,\"mounts\":[{\"name\":\"p1\","
               "\"peer_key\":\"1\",\"space_id\":\"beta\","
               "\"mode\":\"pinned\",\"pinned_root_ref\":\"%s\"}]}",
               pinned_hex);
  free(pinned_hex);
  amduat_reference_free(&pinned_ref);
  if (n <= 0 || (size_t)n >= sizeof(json)) {
    fprintf(stderr, "failed to format manifest json\n");
    free(root);
    return 1;
  }

  if (!amduatd_test_seed_manifest(&store, &pointer_store, &space, json)) {
    fprintf(stderr, "failed to seed manifest\n");
    free(root);
    return 1;
  }

  memset(&transport_state, 0, sizeof(transport_state));
  memset(&transport, 0, sizeof(transport));
  transport.ctx = &transport_state;
  transport.get_records = amduatd_test_get_records;
  transport.free_records = amduatd_test_free_records;
  transport.get_artifact = amduatd_test_get_artifact;

  status = amduatd_space_mounts_sync_until(&store,
                                           &pointer_store,
                                           &space,
                                           &fed_cfg,
                                           &transport,
                                           128u,
                                           10u,
                                           32u,
                                           &report);
  expect(status == AMDUATD_SPACE_MOUNTS_SYNC_OK, "sync no-track ok");
  expect(report.mounts_total == 0u, "no-track total");
  expect(report.mounts_synced == 0u, "no-track synced");
  expect(report.ok, "no-track ok flag");
  expect(report.results_json != NULL &&
             strcmp(report.results_json, "[]") == 0,
         "no-track results empty");
  amduatd_space_mounts_sync_report_free(&report);
  free(root);
  return failures == 0 ? 0 : 1;
}

static int amduatd_test_sync_two_mounts(void) {
  char *root = amduatd_test_make_temp_dir();
  amduat_asl_store_fs_config_t cfg;
  amduatd_store_ctx_t store_ctx;
  amduat_asl_store_t store;
  amduat_asl_pointer_store_t pointer_store;
  amduatd_space_t space;
  amduatd_fed_cfg_t fed_cfg;
  amduatd_test_sync_transport_t transport_state;
  amduatd_fed_pull_transport_t transport;
  const char *json =
      "{\"version\":1,\"mounts\":["
      "{\"name\":\"b\",\"peer_key\":\"2\",\"space_id\":\"beta\",\"mode\":\"track\"},"
      "{\"name\":\"a\",\"peer_key\":\"1\",\"space_id\":\"alpha\",\"mode\":\"track\"}"
      "]}";
  amduatd_space_mounts_sync_report_t report;
  amduatd_space_mounts_sync_status_t status;
  const char *first_a = NULL;
  const char *second_b = NULL;

  if (root == NULL) {
    return 1;
  }
  memset(&cfg, 0, sizeof(cfg));
  if (!amduat_asl_store_fs_init_root(root, NULL, &cfg)) {
    fprintf(stderr, "failed to init store root\n");
    free(root);
    return 1;
  }
  memset(&store_ctx, 0, sizeof(store_ctx));
  memset(&store, 0, sizeof(store));
  if (!amduatd_store_init(&store,
                          &cfg,
                          &store_ctx,
                          root,
                          AMDUATD_STORE_BACKEND_INDEX)) {
    fprintf(stderr, "failed to init store\n");
    free(root);
    return 1;
  }
  if (!amduat_asl_pointer_store_init(&pointer_store, root)) {
    fprintf(stderr, "failed to init pointer store\n");
    free(root);
    return 1;
  }
  if (!amduatd_space_init(&space, "alpha", false)) {
    fprintf(stderr, "failed to init space\n");
    free(root);
    return 1;
  }
  amduatd_fed_cfg_init(&fed_cfg);
  fed_cfg.enabled = true;

  if (!amduatd_test_seed_manifest(&store, &pointer_store, &space, json)) {
    fprintf(stderr, "failed to seed manifest\n");
    free(root);
    return 1;
  }
  if (!amduatd_test_seed_cursor(&store,
                                &pointer_store,
                                &space,
                                "1",
                                "alpha",
                                5u,
                                0x22) ||
      !amduatd_test_seed_cursor(&store,
                                &pointer_store,
                                &space,
                                "2",
                                "beta",
                                7u,
                                0x33)) {
    fprintf(stderr, "failed to seed cursors\n");
    free(root);
    return 1;
  }

  memset(&transport_state, 0, sizeof(transport_state));
  memset(&transport, 0, sizeof(transport));
  transport.ctx = &transport_state;
  transport.get_records = amduatd_test_get_records;
  transport.free_records = amduatd_test_free_records;
  transport.get_artifact = amduatd_test_get_artifact;

  status = amduatd_space_mounts_sync_until(&store,
                                           &pointer_store,
                                           &space,
                                           &fed_cfg,
                                           &transport,
                                           128u,
                                           1u,
                                           32u,
                                           &report);
  expect(status == AMDUATD_SPACE_MOUNTS_SYNC_OK, "sync two ok");
  expect(report.mounts_total == 2u, "sync two total");
  expect(report.mounts_synced == 2u, "sync two synced");
  expect(report.ok, "sync two ok flag");
  expect(transport_state.calls == 2u, "sync two transport calls");
  expect(report.results_json != NULL, "sync two results present");
  if (report.results_json != NULL) {
    first_a = strstr(report.results_json, "\"name\":\"a\"");
    second_b = strstr(report.results_json, "\"name\":\"b\"");
  }
  expect(first_a != NULL && second_b != NULL && first_a < second_b,
         "sync results canonical order");
  expect(strstr(report.results_json, "\"last_logseq\":5") != NULL,
         "sync cursor logseq a");
  expect(strstr(report.results_json, "\"last_logseq\":7") != NULL,
         "sync cursor logseq b");
  amduatd_space_mounts_sync_report_free(&report);
  free(root);
  return failures == 0 ? 0 : 1;
}

static int amduatd_test_sync_partial_error(void) {
  char *root = amduatd_test_make_temp_dir();
  amduat_asl_store_fs_config_t cfg;
  amduatd_store_ctx_t store_ctx;
  amduat_asl_store_t store;
  amduat_asl_pointer_store_t pointer_store;
  amduatd_space_t space;
  amduatd_fed_cfg_t fed_cfg;
  amduatd_test_sync_transport_t transport_state;
  amduatd_fed_pull_transport_t transport;
  const char *json =
      "{\"version\":1,\"mounts\":["
      "{\"name\":\"a\",\"peer_key\":\"peer-a\",\"space_id\":\"alpha\","
      "\"mode\":\"track\"},"
      "{\"name\":\"b\",\"peer_key\":\"2\",\"space_id\":\"beta\",\"mode\":\"track\"}"
      "]}";
  amduatd_space_mounts_sync_report_t report;
  amduatd_space_mounts_sync_status_t status;

  if (root == NULL) {
    return 1;
  }
  memset(&cfg, 0, sizeof(cfg));
  if (!amduat_asl_store_fs_init_root(root, NULL, &cfg)) {
    fprintf(stderr, "failed to init store root\n");
    free(root);
    return 1;
  }
  memset(&store_ctx, 0, sizeof(store_ctx));
  memset(&store, 0, sizeof(store));
  if (!amduatd_store_init(&store,
                          &cfg,
                          &store_ctx,
                          root,
                          AMDUATD_STORE_BACKEND_INDEX)) {
    fprintf(stderr, "failed to init store\n");
    free(root);
    return 1;
  }
  if (!amduat_asl_pointer_store_init(&pointer_store, root)) {
    fprintf(stderr, "failed to init pointer store\n");
    free(root);
    return 1;
  }
  if (!amduatd_space_init(&space, "alpha", false)) {
    fprintf(stderr, "failed to init space\n");
    free(root);
    return 1;
  }
  amduatd_fed_cfg_init(&fed_cfg);
  fed_cfg.enabled = true;

  if (!amduatd_test_seed_manifest(&store, &pointer_store, &space, json)) {
    fprintf(stderr, "failed to seed manifest\n");
    free(root);
    return 1;
  }

  memset(&transport_state, 0, sizeof(transport_state));
  memset(&transport, 0, sizeof(transport));
  transport.ctx = &transport_state;
  transport.get_records = amduatd_test_get_records;
  transport.free_records = amduatd_test_free_records;
  transport.get_artifact = amduatd_test_get_artifact;

  status = amduatd_space_mounts_sync_until(&store,
                                           &pointer_store,
                                           &space,
                                           &fed_cfg,
                                           &transport,
                                           128u,
                                           1u,
                                           32u,
                                           &report);
  expect(status == AMDUATD_SPACE_MOUNTS_SYNC_OK, "sync partial ok");
  expect(!report.ok, "sync partial ok flag false");
  expect(report.mounts_total == 2u, "sync partial total");
  expect(report.mounts_synced == 2u, "sync partial synced");
  expect(transport_state.calls == 1u, "sync partial transport calls");
  expect(report.results_json != NULL &&
             strstr(report.results_json, "\"code\":\"invalid_peer\"") != NULL,
         "sync partial invalid peer");
  amduatd_space_mounts_sync_report_free(&report);
  free(root);
  return failures == 0 ? 0 : 1;
}

int main(void) {
  if (amduatd_test_sync_missing_manifest() != 0) {
    return 1;
  }
  if (amduatd_test_sync_no_track_mounts() != 0) {
    return 1;
  }
  if (amduatd_test_sync_two_mounts() != 0) {
    return 1;
  }
  if (amduatd_test_sync_partial_error() != 0) {
    return 1;
  }
  return failures == 0 ? 0 : 1;
}