apogee/src/controller/inbox.c

#include "inbox.h"

// Submodules
#include "http_server/http_request.h"
#include "json/json.h"
#include "ffdb/fs_list.h"

// Model
#include "model/server.h"
#include "model/account.h"
#include "model/ap/activity.h"
#include "model/ap/inbox_envelope.h"
#include "model/crypto/http_sign.h"

// Stdlib
#include <stdio.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

extern bool terminate;

bool route_inbox( struct http_request* req )
{
	// No subroutes
	if( !http_request_route_term( req, "" ) ) { return false; }
	if( !http_request_route_method( req, "POST" ) ) { return false; }

	if( envelope_create_from_request( req ) ) {
		http_request_send_headers( req, 200, "text/plain", true );
	} else {
		http_request_send_headers( req, 400, "text/plain", true );
	}

	return true;
}
bool route_undo_activity( struct ap_activity* act )
{
	if( act->object.tag != apaot_activity ) {
		// Don't undo activities that are references
		return false;
	}
	if( !act->object.ptr ) {
		printf( "No object in activity\n" );
		return false;
	}

	switch( act->object.ptr->type ) {
		case apat_follow:
			const char* target = act->object.ptr->object.ref;
			struct account* a = account_from_uri( target );
			if( !a ||  0 != strcmp( a->server, g_server_name ) ) {
				printf( "Unfollow not targeted at local account. Discarding.\n" );
				return true;
			}

			struct account* follower = account_from_uri( act->actor );
			if( !follower ) {
				printf( "Follower account not present local. Active follow not possible. Discarding undo\n" );
				return true;
			}

			printf( "TODO: undo %s following %s\n", act->actor, target );
			return false;
		default:
			printf( "Unhandled object activity type %d in undo\n", act->object.ptr->type );
			return false;
	};

	return false;
}

bool route_follow( struct ap_activity* act )
{
	struct account* follower = NULL;
	bool res = false;
	struct ap_activity* accept = NULL;

	const char* target = act->object.ref;
	struct account* a = account_from_uri( target );

	if( !a ||  0 != strcmp( a->server, g_server_name ) ) {
		printf( "Unfollow not targeted at local account. Discarding.\n" );
		goto success;
	}

	follower = account_from_uri( act->actor );
	if( !follower ) {
		follower = account_fetch_from_uri( act->actor );
	}
	if( !follower ) {
		printf( "Unable to fetch account for %s\n", act->actor );
		goto failed;
	}

	// Create Accept activity
	accept = ap_activity_create_accept(act);

	char filename[512]; snprintf( filename, 512, "data/outbox/%d.json", accept->local_id );
	char tmp_filename[512]; snprintf( tmp_filename, 512, "%s.tmp", filename );
	FILE* f = fopen(tmp_filename,"w");
	fprintf( f, "to: %d\n", follower->id );
	ap_activity_write_to_FILE( accept, f );
	fclose(f);

	rename( tmp_filename, filename );

success:
	res = true;
cleanup:
	ap_activity_free(accept);
	account_free(a);
	account_free(follower);

	exit(0);

	return res;
failed:
	res = false;
	goto cleanup;
}

bool route_activity( struct ap_activity* act )
{
	switch( act->type ) {
		case apat_undo:	return route_undo_activity(act);
		case apat_follow:	return route_follow(act);
		default:
			printf( "Unhandled activity type: %d\n", act->type );
	}
	return false;
}

bool process_one()
{
	// Items requiring cleanup
	struct ap_activity* act = NULL;
	struct ap_envelope* env = NULL;
	bool result = false;

	int tail_pos = fs_list_get("data/inbox/TAIL");
	int head_pos = fs_list_get("data/inbox/HEAD");
	if( head_pos <= tail_pos ) { return false; }

	// We have data to process
	printf( "Inbox has %d items pending processing...\n", (head_pos - tail_pos) );

	int id = tail_pos + 1;

	env = ap_envelope_from_id(id);
	bool step_tail = false;

	if( !env ) {
		printf( "Failed to parse envelope+activity for data/inbox/%d.json\n", id );
		return false;
	}

	// Validate signature
	env->validated = http_signature_validate( env, "post /inbox" );

	// Load activity
	FILE* f = fmemopen( env->body, strlen(env->body), "r" );
	act = ap_activity_from_FILE(f);
	if( !act ) { goto failed; }

	// Discard delete requests
	if( act->type == apat_delete ) {
		step_tail = true;
		goto step;
	}

	if( !env->validated ) { goto failed; }

	printf( "Processing %d\n", id );
	step_tail = route_activity( act );

finished:
	printf( "handled: %c\n", step_tail ? 'T' : 'F' );
	if( step_tail ) {
		fs_list_set( "data/inbox/TAIL", id );
		result = true;
	}

	ap_activity_free(act);
	ap_envelope_free(env);
	return result;
failed:
	result = false;
	goto finished;
step:
	result = true;
	goto finished;
}

bool cleanup_inbox()
{
	int tail_pos = fs_list_get("data/inbox/TAIL");
	int dead_pos = fs_list_get("data/inbox/DEAD");
	if( dead_pos < tail_pos ) {
		char filename[512];
		snprintf( filename, 512, "data/inbox/%d.json", dead_pos + 1 );
		FILE* f = fopen( filename, "r" );
		if( f ) {
			fclose(f);
			if( 0 == remove(filename) ) {
				// File successfully remove, advance
				fs_list_set( "data/inbox/DEAD", dead_pos + 1 );

				return true;
			}
		} else {
			// File already doesn't exist, advance
			fclose(f);
			fs_list_set( "data/inbox/DEAD", dead_pos + 1 );

			return true;
		}
	}
	return false;
}

void process_inbox()
{
	while( !terminate ) {
		bool activity = false;
		activity |= process_one();
		if( !activity ) {
			printf( "TODO: unhandled activity\n" );
			exit(1);
		}
		activity |= cleanup_inbox();

		if( !activity ) {
			fflush(stdout);
			sleep(10);
		}
	}
}