/*
 * pgstatheap
 *
 */

#include "postgres.h"

#include "fmgr.h"
#include "funcapi.h"
#include "access/heapam.h"
#include "access/transam.h"
#include "catalog/namespace.h"
#include "catalog/pg_type.h"
#include "storage/bufpage.h"
#include "utils/builtins.h"
#include "utils/inval.h"

PG_FUNCTION_INFO_V1(bufpage_get_raw_page);
PG_FUNCTION_INFO_V1(heap_raw_page_header_is_valid);
PG_FUNCTION_INFO_V1(heap_raw_page_tuple_headers);

extern Datum bufpage_get_raw_page(PG_FUNCTION_ARGS);
extern Datum heap_raw_page_header_is_valid(PG_FUNCTION_ARGS);
extern Datum heap_raw_page_tuple_headers(PG_FUNCTION_ARGS);

static void validate_raw_page(bytea	*raw_page);

#define CHECK_RELATION_BLOCK_RANGE(rel, blkno) { \
		if ( (blkno)<0 && RelationGetNumberOfBlocks((rel))<=(blkno) ) \
			 elog(ERROR, "Block number out of range."); }

#define GET_TEXT(str_) \
        DirectFunctionCall1(textin, CStringGetDatum(str_))

Datum
bufpage_get_raw_page(PG_FUNCTION_ARGS)
{
	text	   *relname = PG_GETARG_TEXT_P(0);
	uint32		blkno = PG_GETARG_UINT32(1);

	Relation	rel;
	RangeVar	*relrv;
	bytea		*raw_page;
	char		*raw_page_data;
	Page		bufpage;
	Buffer		buf;

	relrv = makeRangeVarFromNameList(textToQualifiedNameList(relname));

	rel = relation_openrv(relrv, AccessShareLock);

	/* work on relation */
	{
		CHECK_RELATION_BLOCK_RANGE(rel, blkno);

		buf = ReadBuffer(rel, blkno);

		/* work on buffer */
		{
			bufpage = BufferGetPage(buf);

			raw_page = (bytea *) palloc(BLCKSZ + VARHDRSZ);
			SET_VARSIZE(raw_page, BLCKSZ + VARHDRSZ);
			raw_page_data = VARDATA(raw_page);

			LockBuffer(buf, BUFFER_LOCK_SHARE);

			/* SnapshotEverything */
			{
				memcpy(raw_page_data, bufpage, BLCKSZ);
			}	
			LockBuffer(buf, BUFFER_LOCK_UNLOCK);
		}
		ReleaseBuffer(buf);
	}
	relation_close(rel, AccessShareLock);

	PG_RETURN_BYTEA_P(raw_page);
}

Datum
heap_raw_page_header_is_valid(PG_FUNCTION_ARGS)
{
	bytea	   *raw_page = PG_GETARG_BYTEA_P(0);
	bool		result;

	validate_raw_page(raw_page);
		
	result = PageHeaderIsValid((PageHeader)VARDATA(raw_page));

	PG_RETURN_BOOL(result);
}

static void
validate_raw_page(bytea	*raw_page)
{
	if ((VARSIZE(raw_page) - VARHDRSZ) != BLCKSZ)
		elog(ERROR,"Invalid raw_page: data length %d does not match BLCKSZ", 
						VARSIZE(raw_page) - VARHDRSZ);
}

struct user_args
{
	TupleDesc	tupd;
	Page		page;
	uint16		offset;
};

Datum
heap_raw_page_tuple_headers(PG_FUNCTION_ARGS)
{
	bytea	   *raw_page = PG_GETARG_BYTEA_P(0);

	HeapTuple	tuple;
	ItemId		id;
	Datum		result;

	FuncCallContext *fctx;
	MemoryContext mctx;
	struct user_args *uargs = NULL;

	validate_raw_page(raw_page);

	if (SRF_IS_FIRSTCALL())
	{
		fctx = SRF_FIRSTCALL_INIT();
		mctx = MemoryContextSwitchTo(fctx->multi_call_memory_ctx);

		uargs = palloc(sizeof(struct user_args));

		uargs->tupd = RelationNameGetTupleDesc("public.heap_page_tuple_header_type");
		uargs->offset = FirstOffsetNumber;

		uargs->page = VARDATA(raw_page);

		fctx->max_calls = PageGetMaxOffsetNumber(uargs->page);
		fctx->user_fctx = uargs;

		MemoryContextSwitchTo(mctx);
	}

	fctx = SRF_PERCALL_SETUP();
	uargs = fctx->user_fctx;

	if (fctx->call_cntr < fctx->max_calls)
	{
		bool		id_IsValid;
        Datum		values[12];
        bool		nulls[12];

		id = PageGetItemId(uargs->page, uargs->offset);

		id_IsValid = ItemIdIsValid(id);

		if (id_IsValid)
		{
			HeapTupleData 		tup;
			HeapTupleHeader		tuphdr;
			char				*t_ctid = palloc(16);
			BlockNumber			blkno;

			tuphdr = (HeapTupleHeader) PageGetItem((Page) uargs->page, id);
			tup.t_len = ItemIdGetLength(id);

			blkno = BlockIdGetBlockNumber(&(tuphdr->t_ctid.ip_blkid));

			snprintf(t_ctid, 16, "(%u,%u)", blkno, tuphdr->t_ctid.ip_posid);

			/* ItemOffset */
			values[0] = UInt32GetDatum(uargs->offset);
			nulls[0] = false;
			if (tup.t_len > 0)
			{
				/* Valid? */
				values[1] = BoolGetDatum(true);
				nulls[1] = false;
				/* Length */
				values[2] = UInt32GetDatum(tup.t_len);
				nulls[2] = false;
				/* t_ctid */
				values[3] = GET_TEXT(t_ctid);
				nulls[3] = false;
				/* t_xmin */
				values[4] = UInt32GetDatum(HeapTupleHeaderGetXmin(tuphdr));
				nulls[4] = false;
				/* cn */
				if (tuphdr->t_infomask & HEAP_XMIN_COMMITTED) 
					values[5] = GET_TEXT("c");
				else if (tuphdr->t_infomask & HEAP_XMIN_INVALID)
						values[5] = GET_TEXT("i");
				else
						values[5] = GET_TEXT(" ");
				nulls[5] = false;
				/* t_xmax */
				values[6] = UInt32GetDatum(HeapTupleHeaderGetXmax(tuphdr));
				nulls[6] = false;
				/* cx */
				if (tuphdr->t_infomask & HEAP_XMAX_COMMITTED) 
						values[7] = GET_TEXT("c");
				else if (tuphdr->t_infomask & HEAP_XMAX_INVALID)
						values[7] = GET_TEXT("i");
				else if (tuphdr->t_infomask & HEAP_XMAX_IS_MULTI)
						values[7] = GET_TEXT("m");
				else if (tuphdr->t_infomask & HEAP_XMAX_EXCL_LOCK)
						values[7] = GET_TEXT("x");
				else if (tuphdr->t_infomask & HEAP_XMAX_SHARED_LOCK)
						values[7] = GET_TEXT("s");
				else
						values[7] = GET_TEXT(" ");
				nulls[7] = false;
				/* t_cid */
				values[8] = UInt32GetDatum(HeapTupleHeaderGetRawCommandId(tuphdr));
				nulls[8] = false;
				/* natts */
				values[9] = UInt16GetDatum(HeapTupleHeaderGetNatts(tuphdr));
				nulls[9] = false;
				/* oid */
				values[10] = UInt32GetDatum(HeapTupleHeaderGetOid(tuphdr));
				nulls[10] = false;
				/* info_flag_text */
				if (tuphdr->t_infomask & HEAP_UPDATED) 
					values[11] = GET_TEXT("HEAP_UPDATED");
				else
					values[11] = GET_TEXT(" ");
				nulls[11] = false;
			}
			else
			{
				/* Valid? */
				values[1] = BoolGetDatum(false);
				nulls[1] = false;
				/* Length */
				nulls[2] = true;
				/* t_ctid */
				nulls[3] = true;
				/* t_xmin */
				nulls[4] = true;
				/* cn */
				nulls[5] = true;
				/* t_xmax */
				nulls[6] = true;
				/* cx */
				nulls[7] = true;
				/* t_cid */
				nulls[8] = true;
				/* natts */
				nulls[9] = true;
				/* oid */
				nulls[10] = true;
				/* info_flag_text */
				nulls[11] = true;
			}
		}
		else
		{
			/* ItemOffset */
			values[0] = UInt32GetDatum(uargs->offset);
			nulls[0] = false;
			/* Valid? */
			values[1] = BoolGetDatum(false);
			nulls[1] = false;
			/* Length */
			nulls[2] = true;
			/* t_ctid */
			nulls[3] = true;
			/* t_xmin */
			nulls[4] = true;
			/* cn */
			nulls[5] = true;
			/* t_xmax */
			nulls[6] = true;
			/* cx */
			nulls[7] = true;
			/* t_cid */
			nulls[8] = true;
			/* natts */
			nulls[9] = true;
			/* oid */
			nulls[10] = true;
			/* info_flag_text */
			nulls[11] = true;
		}

        /* Build and return the tuple. */
        tuple = heap_form_tuple(uargs->tupd, values, nulls);
        result = HeapTupleGetDatum(tuple);

		uargs->offset += 1;

		SRF_RETURN_NEXT(fctx, result);
	}
	else
		SRF_RETURN_DONE(fctx);
}