/*-------------------------------------------------------------------------
 *
 * pg_logicalinspect.c
 *    Functions to inspect contents of PostgreSQL logical snapshots
 *
 * Copyright (c) 2024-2025, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *    contrib/pg_logicalinspect/pg_logicalinspect.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "funcapi.h"
#include "replication/snapbuild_internal.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/pg_lsn.h"

PG_MODULE_MAGIC_EXT(
     .name = "pg_logicalinspect",
     .version = PG_VERSION
);

PG_FUNCTION_INFO_V1(pg_get_logical_snapshot_meta);
PG_FUNCTION_INFO_V1(pg_get_logical_snapshot_info);

/* Return the description of SnapBuildState */
static const char *
get_snapbuild_state_desc(SnapBuildState state)
{
 const char *stateDesc = "unknown state";

 switch (state)
 {
  case SNAPBUILD_START:
   stateDesc = "start";
   break;
  case SNAPBUILD_BUILDING_SNAPSHOT:
   stateDesc = "building";
   break;
  case SNAPBUILD_FULL_SNAPSHOT:
   stateDesc = "full";
   break;
  case SNAPBUILD_CONSISTENT:
   stateDesc = "consistent";
   break;
 }

 return stateDesc;
}

/*
 * Extract the LSN from the given serialized snapshot file name.
 */
static XLogRecPtr
parse_snapshot_filename(const char *filename)
{
 uint32  hi;
 uint32  lo;
 XLogRecPtr lsn;
 char  tmpfname[MAXPGPATH];

 /*
  * Extract the values to build the LSN.
  *
  * Note: Including ".snap" doesn't mean that sscanf() also does the format
  * check including the suffix. The subsequent check validates if the given
  * filename has the expected suffix.
 */
 if (sscanf(filename, "%X-%X.snap", &hi, &lo) != 2)
  goto parse_error;

 /*
  * Bring back the extracted LSN to the snapshot file format and compare it
  * to the given filename. This check strictly checks if the given filename
  * follows the format of the snapshot filename.
 */
 sprintf(tmpfname, "%X-%X.snap", hi, lo);
 if (strcmp(tmpfname, filename) != 0)
  goto parse_error;

 lsn = ((uint64) hi) << 32 | lo;

 return lsn;

parse_error:
 ereport(ERROR,
   errmsg("invalid snapshot file name \"%s\"", filename));

 return InvalidXLogRecPtr; /* keep compiler quiet */
}

/*
 * Retrieve the logical snapshot file metadata.
 */
Datum
pg_get_logical_snapshot_meta(PG_FUNCTION_ARGS)
{
#define PG_GET_LOGICAL_SNAPSHOT_META_COLS 3
 SnapBuildOnDisk ondisk;
 HeapTuple tuple;
 Datum  values[PG_GET_LOGICAL_SNAPSHOT_META_COLS] = {0};
 bool  nulls[PG_GET_LOGICAL_SNAPSHOT_META_COLS] = {0};
 TupleDesc tupdesc;
 XLogRecPtr lsn;
 int   i = 0;
 text    *filename_t = PG_GETARG_TEXT_PP(0);

 /* Build a tuple descriptor for our result type */
 if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
  elog(ERROR, "return type must be a row type");

 lsn = parse_snapshot_filename(text_to_cstring(filename_t));

 /* Validate and restore the snapshot to 'ondisk' */
 SnapBuildRestoreSnapshot(&ondisk, lsn, CurrentMemoryContext, false);

 values[i++] = UInt32GetDatum(ondisk.magic);
 values[i++] = Int64GetDatum((int64) ondisk.checksum);
 values[i++] = UInt32GetDatum(ondisk.version);

 Assert(i == PG_GET_LOGICAL_SNAPSHOT_META_COLS);

 tuple = heap_form_tuple(tupdesc, values, nulls);

 PG_RETURN_DATUM(HeapTupleGetDatum(tuple));

#undef PG_GET_LOGICAL_SNAPSHOT_META_COLS
}

Datum
pg_get_logical_snapshot_info(PG_FUNCTION_ARGS)
{
#define PG_GET_LOGICAL_SNAPSHOT_INFO_COLS 14
 SnapBuildOnDisk ondisk;
 HeapTuple tuple;
 Datum  values[PG_GET_LOGICAL_SNAPSHOT_INFO_COLS] = {0};
 bool  nulls[PG_GET_LOGICAL_SNAPSHOT_INFO_COLS] = {0};
 TupleDesc tupdesc;
 XLogRecPtr lsn;
 int   i = 0;
 text    *filename_t = PG_GETARG_TEXT_PP(0);

 /* Build a tuple descriptor for our result type */
 if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
  elog(ERROR, "return type must be a row type");

 lsn = parse_snapshot_filename(text_to_cstring(filename_t));

 /* Validate and restore the snapshot to 'ondisk' */
 SnapBuildRestoreSnapshot(&ondisk, lsn, CurrentMemoryContext, false);

 values[i++] = CStringGetTextDatum(get_snapbuild_state_desc(ondisk.builder.state));
 values[i++] = TransactionIdGetDatum(ondisk.builder.xmin);
 values[i++] = TransactionIdGetDatum(ondisk.builder.xmax);
 values[i++] = LSNGetDatum(ondisk.builder.start_decoding_at);
 values[i++] = LSNGetDatum(ondisk.builder.two_phase_at);
 values[i++] = TransactionIdGetDatum(ondisk.builder.initial_xmin_horizon);
 values[i++] = BoolGetDatum(ondisk.builder.building_full_snapshot);
 values[i++] = BoolGetDatum(ondisk.builder.in_slot_creation);
 values[i++] = LSNGetDatum(ondisk.builder.last_serialized_snapshot);
 values[i++] = TransactionIdGetDatum(ondisk.builder.next_phase_at);

 values[i++] = UInt32GetDatum(ondisk.builder.committed.xcnt);
 if (ondisk.builder.committed.xcnt > 0)
 {
  Datum    *arrayelems;

  arrayelems = (Datum *) palloc(ondisk.builder.committed.xcnt * sizeof(Datum));

  for (int j = 0; j < ondisk.builder.committed.xcnt; j++)
   arrayelems[j] = TransactionIdGetDatum(ondisk.builder.committed.xip[j]);

  values[i++] = PointerGetDatum(construct_array_builtin(arrayelems,
                 ondisk.builder.committed.xcnt,
                 XIDOID));
 }
 else
  nulls[i++] = true;

 values[i++] = UInt32GetDatum(ondisk.builder.catchange.xcnt);
 if (ondisk.builder.catchange.xcnt > 0)
 {
  Datum    *arrayelems;

  arrayelems = (Datum *) palloc(ondisk.builder.catchange.xcnt * sizeof(Datum));

  for (int j = 0; j < ondisk.builder.catchange.xcnt; j++)
   arrayelems[j] = TransactionIdGetDatum(ondisk.builder.catchange.xip[j]);

  values[i++] = PointerGetDatum(construct_array_builtin(arrayelems,
                 ondisk.builder.catchange.xcnt,
                 XIDOID));
 }
 else
  nulls[i++] = true;

 Assert(i == PG_GET_LOGICAL_SNAPSHOT_INFO_COLS);

 tuple = heap_form_tuple(tupdesc, values, nulls);

 PG_RETURN_DATUM(HeapTupleGetDatum(tuple));

#undef PG_GET_LOGICAL_SNAPSHOT_INFO_COLS
}