From c95d1e56ee232bcbba7581ff226385871cd8b5d2 Mon Sep 17 00:00:00 2001
From: Jason Volk <jason@zemos.net>
Date: Tue, 13 Aug 2019 22:59:27 -0700
Subject: [PATCH] ircd::m::init: Add preliminary worker context loop for
 backfill unit. (#78)

---
 include/ircd/m/m.h         |   1 +
 ircd/m.cc                  |   1 +
 modules/Makefile.am        |   2 +
 modules/m_init_backfill.cc | 140 +++++++++++++++++++++++++++++++++++++
 4 files changed, 144 insertions(+)
 create mode 100644 modules/m_init_backfill.cc

diff --git a/include/ircd/m/m.h b/include/ircd/m/m.h
index 7fd845fff..abbdf9bf2 100644
--- a/include/ircd/m/m.h
+++ b/include/ircd/m/m.h
@@ -83,6 +83,7 @@ namespace ircd::m
 struct ircd::m::init
 {
 	struct modules;
+	struct backfill;
 
 	self::init _self;
 	dbs::init _dbs;
diff --git a/ircd/m.cc b/ircd/m.cc
index 61dd94052..366c5b483 100644
--- a/ircd/m.cc
+++ b/ircd/m.cc
@@ -320,6 +320,7 @@ ircd::m::module_names
 	"web_hook",
 	"m_listen",
 	"m_vm",
+	"m_init_backfill",
 	"stats",
 };
 
diff --git a/modules/Makefile.am b/modules/Makefile.am
index f99f551ec..36dc69ec0 100644
--- a/modules/Makefile.am
+++ b/modules/Makefile.am
@@ -138,6 +138,7 @@ m_room_third_party_invite_la_SOURCES = m_room_third_party_invite.cc
 m_room_redaction_la_SOURCES = m_room_redaction.cc
 m_room_bootstrap_la_SOURCES = m_room_bootstrap.cc
 m_init_bootstrap_la_SOURCES = m_init_bootstrap.cc
+m_init_backfill_la_SOURCES = m_init_backfill.cc
 m_listen_la_SOURCES = m_listen.cc
 m_users_la_SOURCES = m_users.cc
 
@@ -182,6 +183,7 @@ m_module_LTLIBRARIES = \
 	m_room_server_acl.la \
 	m_room_bootstrap.la \
 	m_init_bootstrap.la \
+	m_init_backfill.la \
 	m_listen.la \
 	m_users.la \
 	###
diff --git a/modules/m_init_backfill.cc b/modules/m_init_backfill.cc
new file mode 100644
index 000000000..8b208c27d
--- /dev/null
+++ b/modules/m_init_backfill.cc
@@ -0,0 +1,140 @@
+// Matrix Construct
+//
+// Copyright (C) Matrix Construct Developers, Authors & Contributors
+// Copyright (C) 2016-2019 Jason Volk <jason@zemos.net>
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice is present in all copies. The
+// full license for this software is available in the LICENSE file.
+
+/// This should be a namespace but we're stuck in struct m::init for now, so
+/// this code should be portable for a future when m::init is unstructured.
+struct ircd::m::init::backfill
+{
+	static void handle_room(const room::id &);
+	static void worker();
+	static void fini();
+	static void init();
+
+	static std::unique_ptr<context> worker_context;
+	static log::log log;
+};
+
+ircd::mapi::header
+IRCD_MODULE
+{
+	"Matrix resynchronization backfilling",
+	ircd::m::init::backfill::init,
+	ircd::m::init::backfill::fini,
+};
+
+decltype(ircd::m::init::backfill::log)
+ircd::m::init::backfill::log
+{
+	"m.init.backfill"
+};
+
+decltype(ircd::m::init::backfill::worker_context)
+ircd::m::init::backfill::worker_context;
+
+void
+ircd::m::init::backfill::init()
+{
+	assert(!worker_context);
+	worker_context.reset(new context
+	{
+		"m.init.backfill",
+		512_KiB,
+		&worker,
+		context::POST
+	});
+}
+
+void
+ircd::m::init::backfill::fini()
+{
+	worker_context.reset(nullptr);
+}
+
+void
+ircd::m::init::backfill::worker()
+try
+{
+	// The common case is that we're in runlevel START when this context is
+	// entered; we don't want to start this operation until we're in RUN.
+	run::changed::dock.wait([]
+	{
+		return run::level != run::level::START;
+	});
+
+	// If some other level is observed here we shouldn't run this operation.
+	if(run::level != run::level::RUN)
+		return;
+
+	log::debug
+	{
+		log, "Starting initial resynchronization from other servers..."
+	};
+
+	// Iterate all of the rooms this server is aware of which contain
+	// at least one user from another server which is joined to the room.
+	m::rooms::opts opts;
+	opts.remote_joined_only = true;
+	m::rooms::for_each(opts, []
+	(const m::room::id &room_id)
+	{
+		handle_room(room_id);
+		return true;
+	});
+}
+catch(const ctx::interrupted &e)
+{
+	log::derror
+	{
+		log, "Worker interrupted :%s",
+		e.what(),
+	};
+
+	throw;
+}
+catch(const std::exception &e)
+{
+	log::critical
+	{
+		log, "Worker fatal :%s",
+		e.what(),
+	};
+}
+
+void
+ircd::m::init::backfill::handle_room(const room::id &room_id)
+try
+{
+	const m::room room
+	{
+		room_id
+	};
+
+	const m::room::origins origins
+	{
+		room
+	};
+
+	log::debug
+	{
+		log, "Resynchronizing %s with %zu joined servers.",
+		string_view{room_id},
+		origins.count(),
+	};
+
+}
+catch(const std::exception &e)
+{
+	log::error
+	{
+		log, "Failed to synchronize %s :%s",
+		string_view{room_id},
+		e.what(),
+	};
+}