No OneTemporary
Actions

Size

47 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/src/client/actions/sync.cpp b/src/client/actions/sync.cpp
	index 8eefdbe..6565a1b 100644
	--- a/src/client/actions/sync.cpp
	+++ b/src/client/actions/sync.cpp
	@@ -1,330 +1,335 @@
	/*
	* Copyright (C) 2020 Tusooa Zhu
	*
	* This file is part of libkazv.
	*
	* libkazv is free software: you can redistribute it and/or modify
	* it under the terms of the GNU Affero General Public License as
	* published by the Free Software Foundation, either version 3 of the
	* License, or (at your option) any later version.
	*
	* libkazv is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	* GNU Affero General Public License for more details.
	*
	* You should have received a copy of the GNU Affero General Public License
	* along with libkazv. If not, see <https://www.gnu.org/licenses/>.
	*/

	#include <libkazv-config.hpp>

	#include <lager/util.hpp>
	#include <zug/transducer/map.hpp>
	#include <zug/transducer/cat.hpp>
	#include <zug/transducer/filter.hpp>
	#include <zug/sequence.hpp>

	#include <jobinterface.hpp>
	#include <debug.hpp>

	#include "cursorutil.hpp"

	#include "sync.hpp"

	#include "encryption.hpp"

	namespace Kazv
	{
	// Atomicity guaranteed: if the sync action is created
	// before an action that reasonably changes Client
	// (e.g. roll back to an earlier state, obtain other
	// events), but executed
	// after that action, the sync will still give continuous
	// data about the events. (Sync will not "skip" events)
	// This is because this function takes the sync token
	// from the ClientModel model it is passed.
	ClientResult updateClient(ClientModel m, SyncAction)
	{
	kzo.client.dbg() << "Start syncing with token " <<
	(m.syncToken ? m.syncToken.value() : "<null>") << std::endl;

	bool isInitialSync = ! m.syncToken;

	std::string filter = m.syncToken ? m.incrementalSyncFilterId : m.initialSyncFilterId;
	m.addJob(m.job<SyncJob>()
	- .make(filter, m.syncToken)
	+ .make(filter,
	+ m.syncToken,
	+ std::nullopt, // fullState
	+ std::nullopt, // setPresence
	+ // Let initial sync return immediately
	+ isInitialSync ? 0 : m.syncTimeoutMs
	+ )
	.withData(json{{"is", isInitialSync ? "initial" : "incremental"}}));
	return { m, lager::noop };
	}


	static KazvEventList loadRoomsFromSyncInPlace(ClientModel &m, SyncJob::Rooms rooms)
	{
	auto l = std::move(m.roomList);
	auto eventsToEmit = KazvEventList{}.transient();

	auto updateRoomImpl =
	[&l](auto id, auto a) {
	l = RoomListModel::update(
	std::move(l),
	UpdateRoomAction{std::move(id), std::move(a)});
	};
	auto updateSingleRoom =
	[&, updateRoomImpl](const auto &id, const auto &room, auto membership) {
	if (!l.has(id) \|\| l[id].membership != membership) {
	eventsToEmit.push_back(RoomMembershipChanged{membership, id});
	}
	updateRoomImpl(id, ChangeMembershipAction{membership});

	auto timelineEvents =
	intoImmer(
	EventList{},
	zug::map([=](Event e) {
	return Event::fromSync(e, id);
	}),
	room.timeline.events);
	eventsToEmit.append(
	intoImmer(
	KazvEventList{},
	zug::map([=](Event e) -> KazvEvent {
	return ReceivingRoomTimelineEvent{std::move(e), id};
	}),
	timelineEvents).transient());
	updateRoomImpl(id, AppendTimelineAction{timelineEvents});
	if (room.state) {
	eventsToEmit.append(
	intoImmer(
	KazvEventList{},
	zug::map([=](Event e) -> KazvEvent {
	return ReceivingRoomStateEvent{std::move(e), id};
	}),
	room.state.value().events).transient());
	updateRoomImpl(id, AddStateEventsAction{room.state.value().events});
	}
	if (room.accountData) {
	eventsToEmit.append(
	intoImmer(
	KazvEventList{},
	zug::map([=](Event e) -> KazvEvent {
	return ReceivingRoomAccountDataEvent{std::move(e), id};
	}),
	room.state.value().events).transient());
	updateRoomImpl(id, AddAccountDataAction{room.accountData.value().events});
	}
	};

	auto updateJoinedRoom =
	[=](const auto &id, const auto &room) {
	updateSingleRoom(id, room, RoomMembership::Join);
	if (room.ephemeral) {
	updateRoomImpl(id, AddEphemeralAction{room.ephemeral.value().events});
	}
	// TODO update other info such as
	// notification and summary
	};

	auto updateInvitedRoom =
	[=](const auto &id, const auto &room) {
	updateRoomImpl(id, ChangeMembershipAction{RoomMembership::Invite});
	if (room.inviteState) {
	updateRoomImpl(id, ChangeInviteStateAction{room.inviteState.value().events});
	}
	};

	auto updateLeftRoom =
	[=](const auto &id, const auto &room) {
	updateSingleRoom(id, room, RoomMembership::Leave);
	};

	for (const auto &[id, room]: rooms.join) {
	updateJoinedRoom(id, room);
	}

	// TODO update info for invited rooms
	for (const auto &[id, room]: rooms.invite) {
	updateInvitedRoom(id, room);
	}

	for (const auto &[id, room]: rooms.leave) {
	updateLeftRoom(id, room);
	}

	m.roomList = std::move(l);
	return eventsToEmit.persistent();
	}

	static KazvEventList loadPresenceFromSyncInPlace(ClientModel &m, EventList presence)
	{
	auto eventsToEmit = intoImmer(
	KazvEventList{},
	zug::map([](Event e) { return ReceivingPresenceEvent{e}; }),
	presence);
	m.presence = merge(std::move(m.presence), presence, keyOfPresence);
	return eventsToEmit;
	}

	static KazvEventList loadAccountDataFromSyncInPlace(ClientModel &m, EventList accountData)
	{
	auto eventsToEmit = intoImmer(
	KazvEventList{},
	zug::map([](Event e) { return ReceivingPresenceEvent{e}; }),
	accountData);
	m.accountData = merge(std::move(m.accountData), accountData, keyOfAccountData);
	return eventsToEmit;
	}

	static KazvEventList loadToDeviceFromSyncInPlace(ClientModel &m, JsonWrap toDevice)
	{
	if (toDevice.get().contains("events")) {
	auto events = toDevice.get()["events"];
	auto msgs = intoImmer(
	EventList{},
	zug::map([](const json &j) { return Event(j); }),
	events);

	m.toDevice = std::move(m.toDevice) + msgs;

	return intoImmer(
	KazvEventList{},
	zug::map([](Event e) { return ReceivingToDeviceMessage{e}; }),
	msgs);
	}
	return {};
	}

	ClientResult processResponse(ClientModel m, SyncResponse r)
	{
	if (! r.success()) {
	m.addTrigger(SyncFailed{});
	m.syncing = false;
	kzo.client.dbg() << "Sync failed" << std::endl;
	kzo.client.dbg() << r.statusCode << std::endl;
	if (isBodyJson(r.body)) {
	auto j = r.jsonBody();
	kzo.client.dbg() << "Json says: " << j.get().dump() << std::endl;
	} else {
	kzo.client.dbg() << "Response body: "
	<< std::get<BaseJob::BytesBody>(r.body) << std::endl;
	}
	return { std::move(m), lager::noop };
	}

	kzo.client.dbg() << "Sync successful" << std::endl;

	auto rooms = r.rooms();
	auto accountData = r.accountData();
	auto presence = r.presence();
	// load the info that has been sync'd

	m.syncToken = r.nextBatch();

	if (rooms) {
	m.addTriggers(loadRoomsFromSyncInPlace(m, std::move(rooms.value())));
	}

	if (presence) {
	m.addTriggers(loadPresenceFromSyncInPlace(m, std::move(presence.value().events)));
	}

	if (accountData) {
	m.addTriggers(loadAccountDataFromSyncInPlace(m, std::move(accountData.value().events)));
	}

	m.addTriggers(loadToDeviceFromSyncInPlace(m, r.toDevice()));

	auto is = r.dataStr("is");
	auto isInitialSync = is == "initial";

	if (m.crypto) {
	kzo.client.dbg() << "E2EE is on. Processing device lists and one-time key counts." << std::endl;
	auto &crypto = m.crypto.value();
	// process deviceLists
	if (isInitialSync) {
	auto encryptedUsers =
	zug::sequence(
	zug::map([](auto n) { return n.second; })
	\| zug::filter([](auto room) { return room.encrypted; })
	\| zug::map([](auto room) { return room.joinedMemberIds(); })
	\| zug::cat,
	// no need to use distinct here as the map will overwrite
	m.roomList.rooms);

	m.deviceLists.track(std::move(encryptedUsers));

	} else {
	const auto &l = r.deviceLists().get();
	if (l.contains("changed")) {
	const auto &changed = l.at("changed");
	m.deviceLists.track(changed);
	}

	if (l.contains("left")) {
	const auto &left = l.at("left");
	m.deviceLists.untrack(left);
	}
	}

	// deviceOneTimeKeysCount
	crypto.setUploadedOneTimeKeysCount(r.deviceOneTimeKeysCount());

	auto model = tryDecryptEvents(std::move(m));
	m = std::move(model);
	}

	m.addTrigger(SyncSuccessful{r.nextBatch()});
	- m.addTrigger(ShouldQueryKeys{isInitialSync});

	return { std::move(m), lager::noop };
	}

	ClientResult updateClient(ClientModel m, PostInitialFiltersAction)
	{
	if (m.syncing) {
	return { std::move(m), lager::noop };
	}

	Filter initialSyncFilter;
	initialSyncFilter.room.timeline.limit = 1;
	initialSyncFilter.room.state.lazyLoadMembers = true;
	auto firstJob = m.job<DefineFilterJob>()
	.make(m.userId, initialSyncFilter)
	.withData(json{{"is", "initialSyncFilter"}})
	.withQueue("post-filter", CancelFutureIfFailed);
	kzo.client.dbg() << "First filter: " << firstJob.requestBody() << std::endl;
	m.addJob(firstJob);
	Filter incrementalSyncFilter;
	incrementalSyncFilter.room.timeline.limit = 20;
	incrementalSyncFilter.room.state.lazyLoadMembers = true;
	m.addJob(m.job<DefineFilterJob>()
	.make(m.userId, incrementalSyncFilter)
	.withData(json{{"is", "incrementalSyncFilter"}})
	.withQueue("post-filter", CancelFutureIfFailed));

	m.syncing = true;

	return { std::move(m), lager::noop };
	}

	ClientResult processResponse(ClientModel m, DefineFilterResponse r)
	{
	auto is = r.dataStr("is");
	if (! r.success()) {
	m.syncing = false;
	kzo.client.dbg() << "posting filter failed: " << r.errorCode() << r.errorMessage() << std::endl;
	m.addTrigger(PostInitialFiltersFailed{r.errorCode(), r.errorMessage()});
	return { std::move(m), lager::noop };
	}

	kzo.client.dbg() << "filter " << is << " is posted" << std::endl;

	if (is == "incrementalSyncFilter") {
	m.incrementalSyncFilterId = r.filterId();
	m.addTrigger(PostInitialFiltersSuccessful{});
	} else {
	m.initialSyncFilterId = r.filterId();
	}

	return { std::move(m), lager::noop };
	}
	}
	diff --git a/src/client/client-model.hpp b/src/client/client-model.hpp
	index f0b97ff..be96f69 100644
	--- a/src/client/client-model.hpp
	+++ b/src/client/client-model.hpp
	@@ -1,384 +1,385 @@
	/*
	* Copyright (C) 2020 Tusooa Zhu
	*
	* This file is part of libkazv.
	*
	* libkazv is free software: you can redistribute it and/or modify
	* it under the terms of the GNU Affero General Public License as
	* published by the Free Software Foundation, either version 3 of the
	* License, or (at your option) any later version.
	*
	* libkazv is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	* GNU Affero General Public License for more details.
	*
	* You should have received a copy of the GNU Affero General Public License
	* along with libkazv. If not, see <https://www.gnu.org/licenses/>.
	*/


	#pragma once
	#include <libkazv-config.hpp>

	#include <tuple>
	#include <variant>
	#include <string>
	#include <optional>

	#include <lager/context.hpp>
	#include <boost/hana.hpp>

	#ifndef NDEBUG
	#include <lager/debug/cereal/struct.hpp>
	#endif

	#include <csapi/sync.hpp>

	#include <crypto.hpp>

	#include "clientfwd.hpp"
	#include "device-list-tracker.hpp"
	#include "error.hpp"
	#include "room/room-model.hpp"

	namespace Kazv
	{
	inline const std::string DEFTXNID{"0"};

	enum RoomVisibility
	{
	Private,
	Public,
	};

	enum CreateRoomPreset
	{
	PrivateChat,
	PublicChat,
	TrustedPrivateChat,
	};

	enum ThumbnailResizingMethod
	{
	Crop,
	Scale,
	};

	struct ClientModel
	{
	std::string serverUrl;
	std::string userId;
	std::string token;
	std::string deviceId;
	bool loggedIn{false};
	Error error;

	bool syncing{false};
	+ int syncTimeoutMs{20000};
	std::string initialSyncFilterId;
	std::string incrementalSyncFilterId;
	std::optional<std::string> syncToken;

	RoomListModel roomList;
	immer::map<std::string /* sender */, Event> presence;
	immer::map<std::string /* type */, Event> accountData;

	std::string nextTxnId{DEFTXNID};
	immer::flex_vector<BaseJob> nextJobs;
	immer::flex_vector<KazvEvent> nextTriggers;

	EventList toDevice;
	std::optional<Crypto> crypto;
	bool identityKeysUploaded{false};

	DeviceListTracker deviceLists;

	immer::flex_vector<std::string /* deviceId */> devicesToSendKeys(std::string userId) const;

	std::pair<Event, std::optional<std::string> /* sessionKey */> megOlmEncrypt(Event e, std::string roomId);
	/// precondition: the one-time keys for those devices must already be claimed
	Event olmEncrypt(Event e, immer::map<std::string, immer::flex_vector<std::string>> userIdToDeviceIdMap);

	// helpers
	template<class Job>
	struct MakeJobT
	{
	template<class ...Args>
	constexpr auto make(Args &&...args) const {
	if constexpr (Job::needsAuth()) {
	return Job(
	serverUrl,
	token,
	std::forward<Args>(args)...);
	} else {
	return Job(
	serverUrl,
	std::forward<Args>(args)...);
	}
	}

	std::string serverUrl;
	std::string token;
	};

	template<class Job>
	constexpr auto job() const {
	return MakeJobT<Job>{serverUrl, token};
	}

	inline void addJob(BaseJob j) {
	nextJobs = std::move(nextJobs).push_back(std::move(j));
	}

	inline auto popAllJobs() {
	auto jobs = std::move(nextJobs);
	nextJobs = DEFVAL;
	return jobs;
	};

	inline void addTrigger(KazvEvent t) {
	addTriggers({t});
	}

	inline void addTriggers(immer::flex_vector<KazvEvent> c) {
	nextTriggers = std::move(nextTriggers) + c;
	}

	inline auto popAllTriggers() {
	auto triggers = std::move(nextTriggers);
	nextTriggers = DEFVAL;
	return triggers;
	}

	using Action = ClientAction;
	using Effect = ClientEffect;
	using Result = ClientResult;

	static Result update(ClientModel m, Action a);
	};

	// actions:
	struct LoginAction {
	std::string serverUrl;
	std::string username;
	std::string password;
	std::optional<std::string> deviceName;
	};

	struct TokenLoginAction
	{
	std::string serverUrl;
	std::string username;
	std::string token;
	std::string deviceId;
	};

	struct LogoutAction {};

	struct SyncAction {};

	struct PaginateTimelineAction
	{
	std::string roomId;
	std::optional<int> limit;
	};

	struct SendMessageAction
	{
	std::string roomId;
	Event event;
	};

	struct SendStateEventAction
	{
	std::string roomId;
	Event event;
	};

	struct CreateRoomAction
	{
	using Visibility = RoomVisibility;
	using Preset = CreateRoomPreset;
	Visibility visibility;
	std::optional<std::string> roomAliasName;
	std::optional<std::string> name;
	std::optional<std::string> topic;
	immer::array<std::string> invite;
	//immer::array<Invite3pid> invite3pid;
	std::optional<std::string> roomVersion;
	JsonWrap creationContent;
	immer::array<Event> initialState;
	std::optional<Preset> preset;
	std::optional<bool> isDirect;
	JsonWrap powerLevelContentOverride;
	};

	struct GetRoomStatesAction
	{
	std::string roomId;
	};

	struct GetStateEventAction
	{
	std::string roomId;
	std::string type;
	std::string stateKey;
	};

	struct InviteToRoomAction
	{
	std::string roomId;
	std::string userId;
	};

	struct JoinRoomByIdAction
	{
	std::string roomId;
	};

	struct JoinRoomAction
	{
	std::string roomIdOrAlias;
	immer::array<std::string> serverName;
	};

	struct LeaveRoomAction
	{
	std::string roomId;
	};

	struct ForgetRoomAction
	{
	std::string roomId;
	};


	struct SetTypingAction
	{
	std::string roomId;
	bool typing;
	std::optional<int> timeoutMs;
	};

	struct PostReceiptAction
	{
	std::string roomId;
	std::string eventId;
	};

	struct SetReadMarkerAction
	{
	std::string roomId;
	std::string eventId;
	};

	struct UploadContentAction
	{
	immer::box<Bytes> content;
	std::optional<std::string> filename;
	std::optional<std::string> contentType;
	std::string uploadId; // to be used by library users
	};

	struct DownloadContentAction
	{
	std::string mxcUri;
	};

	struct DownloadThumbnailAction
	{
	std::string mxcUri;
	int width;
	int height;
	std::optional<ThumbnailResizingMethod> method;
	std::optional<bool> allowRemote;
	};

	struct ResubmitJobAction
	{
	BaseJob job;
	};

	struct ProcessResponseAction
	{
	Response response;
	};

	struct PostInitialFiltersAction
	{
	};

	struct SendToDeviceMessageAction
	{
	Event event;
	immer::map<std::string, immer::flex_vector<std::string>> devicesToSend;
	};

	struct UploadIdentityKeysAction
	{
	};

	struct GenerateAndUploadOneTimeKeysAction
	{
	};

	struct QueryKeysAction
	{
	bool isInitialSync;
	};

	struct ClaimKeysAndSendSessionKeyAction
	{
	std::string roomId;
	std::string sessionId;
	std::string sessionKey;
	immer::map<std::string, immer::flex_vector<std::string>> devicesToSend;
	};

	#ifndef NDEBUG
	LAGER_CEREAL_STRUCT(LoginAction);
	LAGER_CEREAL_STRUCT(TokenLoginAction);
	LAGER_CEREAL_STRUCT(LogoutAction);
	LAGER_CEREAL_STRUCT(SyncAction);
	LAGER_CEREAL_STRUCT(PostInitialFiltersAction);

	LAGER_CEREAL_STRUCT(PaginateTimelineAction);
	LAGER_CEREAL_STRUCT(SendMessageAction);
	LAGER_CEREAL_STRUCT(SendStateEventAction);
	LAGER_CEREAL_STRUCT(SendToDeviceMessageAction);
	LAGER_CEREAL_STRUCT(CreateRoomAction);
	LAGER_CEREAL_STRUCT(GetRoomStatesAction);
	LAGER_CEREAL_STRUCT(GetStateEventAction);
	LAGER_CEREAL_STRUCT(InviteToRoomAction);
	LAGER_CEREAL_STRUCT(JoinRoomByIdAction);
	LAGER_CEREAL_STRUCT(JoinRoomAction);
	LAGER_CEREAL_STRUCT(LeaveRoomAction);
	LAGER_CEREAL_STRUCT(ForgetRoomAction);
	LAGER_CEREAL_STRUCT(SetTypingAction);
	LAGER_CEREAL_STRUCT(PostReceiptAction);
	LAGER_CEREAL_STRUCT(ProcessResponseAction);
	LAGER_CEREAL_STRUCT(SetReadMarkerAction);

	LAGER_CEREAL_STRUCT(UploadContentAction);
	LAGER_CEREAL_STRUCT(DownloadContentAction);
	LAGER_CEREAL_STRUCT(DownloadThumbnailAction);

	LAGER_CEREAL_STRUCT(ResubmitJobAction);
	#endif

	template<class Archive>
	void serialize(Archive &ar, ClientModel &m, std::uint32_t const /version/)
	{
	ar(m.serverUrl, m.userId, m.token, m.deviceId, m.loggedIn,
	m.error,
	m.initialSyncFilterId,
	m.incrementalSyncFilterId,
	m.syncToken,
	m.roomList,
	m.presence,
	m.accountData,
	m.nextTxnId,
	m.toDevice);
	}
	}
	CEREAL_CLASS_VERSION(Kazv::ClientModel, 0);
	diff --git a/src/client/client.cpp b/src/client/client.cpp
	index 5e6071d..9077a31 100644
	--- a/src/client/client.cpp
	+++ b/src/client/client.cpp
	@@ -1,143 +1,225 @@
	/*
	* Copyright (C) 2020 Tusooa Zhu
	*
	* This file is part of libkazv.
	*
	* libkazv is free software: you can redistribute it and/or modify
	* it under the terms of the GNU Affero General Public License as
	* published by the Free Software Foundation, either version 3 of the
	* License, or (at your option) any later version.
	*
	* libkazv is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	* GNU Affero General Public License for more details.
	*
	* You should have received a copy of the GNU Affero General Public License
	* along with libkazv. If not, see <https://www.gnu.org/licenses/>.
	*/

	#include <libkazv-config.hpp>

	#include <lager/constant.hpp>

	#include "client.hpp"

	namespace Kazv
	{
	Client::Client(lager::reader<SdkModel> sdk,
	Context<ClientAction> ctx)
	: m_sdk(sdk)
	, m_client(sdk.map(&SdkModel::c))
	, m_ctx(std::move(ctx))
	{
	}

	Room Client::room(std::string id) const
	{
	return Room(m_sdk, lager::make_constant(id), m_ctx);
	}

	Room Client::roomByCursor(lager::reader<std::string> id) const
	{
	return Room(m_sdk, id, m_ctx);
	}

	auto Client::passwordLogin(std::string homeserver, std::string username,
	std::string password, std::string deviceName) const
	-> PromiseT
	{
	- return m_ctx.dispatch(LoginAction{
	+ auto p1 = m_ctx.dispatch(LoginAction{
	homeserver, username, password, deviceName});
	+ p1
	+ .then([*this](auto stat) {
	+ if (! stat.success()) {
	+ return;
	+ }
	+ // It is meaningless to wait for it in a Promise
	+ // that is never exposed to the user.
	+ startSyncing();
	+ });
	+
	+ return p1;
	}

	auto Client::tokenLogin(std::string homeserver, std::string username,
	std::string token, std::string deviceId) const
	-> PromiseT
	{
	- return m_ctx.dispatch(TokenLoginAction{
	+ auto p1 = m_ctx.dispatch(TokenLoginAction{
	homeserver, username, token, deviceId});
	+ p1
	+ .then([*this](auto stat) {
	+ if (! stat.success()) {
	+ return;
	+ }
	+ startSyncing();
	+ });
	+
	+ return p1;
	}

	auto Client::createRoom(RoomVisibility v,
	std::optional<std::string> name,
	std::optional<std::string> alias,
	immer::array<std::string> invite,
	std::optional<bool> isDirect,
	bool allowFederate,
	std::optional<std::string> topic,
	JsonWrap powerLevelContentOverride) const
	-> PromiseT
	{
	CreateRoomAction a;
	a.visibility = v;
	a.name = name;
	a.roomAliasName = alias;
	a.invite = invite;
	a.isDirect = isDirect;
	a.topic = topic;
	a.powerLevelContentOverride = powerLevelContentOverride;
	// Synapse won't buy it if we do not provide
	// a creationContent object.
	a.creationContent = json{
	{"m.federate", allowFederate}
	};
	return m_ctx.dispatch(std::move(a));
	}

	auto Client::joinRoomById(std::string roomId) const -> PromiseT
	{
	return m_ctx.dispatch(JoinRoomByIdAction{roomId});
	}

	auto Client::joinRoom(std::string roomId, immer::array<std::string> serverName) const
	-> PromiseT
	{
	return m_ctx.dispatch(JoinRoomAction{roomId, serverName});
	}

	auto Client::uploadContent(immer::box<Bytes> content,
	std::string uploadId,
	std::optional<std::string> filename,
	std::optional<std::string> contentType) const
	-> PromiseT
	{
	return m_ctx.dispatch(UploadContentAction{content, filename, contentType, uploadId});
	}

	auto Client::downloadContent(std::string mxcUri) const
	-> PromiseT
	{
	return m_ctx.dispatch(DownloadContentAction{mxcUri});
	}

	auto Client::downloadThumbnail(
	std::string mxcUri,
	int width,
	int height,
	std::optional<ThumbnailResizingMethod> method) const
	-> PromiseT
	{
	return m_ctx.dispatch(DownloadThumbnailAction{mxcUri, width, height, method, std::nullopt});
	}

	auto Client::startSyncing() const -> PromiseT
	{
	using namespace Kazv::CursorOp;

	if (+syncing()) {
	return m_ctx.createResolvedPromise(true);
	}

	- // filters are incomplete
	- if ((+m_client[&ClientModel::initialSyncFilterId]).empty()
	- \|\| (+m_client[&ClientModel::incrementalSyncFilterId]).empty()) {
	- return m_ctx.dispatch(PostInitialFiltersAction{});
	- } else if (+m_client[&ClientModel::crypto] // encryption is on
	- && ! +m_client[&ClientModel::identityKeysUploaded]) { // but identity keys are not published
	- return m_ctx.dispatch(UploadIdentityKeysAction{});
	- } else { // sync is just interrupted
	- return m_ctx.dispatch(SyncAction{});
	- }
	+ auto ctx = m_ctx;
	+
	+ auto p1 = ctx.createResolvedPromise(true)
	+ .then([=](auto) {
	+ // post filters, if filters are incomplete
	+ if ((+m_client[&ClientModel::initialSyncFilterId]).empty()
	+ \|\| (+m_client[&ClientModel::incrementalSyncFilterId]).empty()) {
	+ return ctx.dispatch(PostInitialFiltersAction{});
	+ }
	+ return ctx.createResolvedPromise(true);
	+ })
	+ .then([=](auto stat) {
	+ if (! stat.success()) {
	+ return ctx.createResolvedPromise(stat);
	+ }
	+ // Upload identity keys if we need to
	+ if (+m_client[&ClientModel::crypto]
	+ && ! +m_client[&ClientModel::identityKeysUploaded]) {
	+ return ctx.dispatch(UploadIdentityKeysAction{});
	+ } else {
	+ return ctx.createResolvedPromise(true);
	+ }
	+ });
	+
	+ p1
	+ .then([*this](auto stat) {
	+ if (stat.success()) {
	+ syncForever();
	+ }
	+ });
	+
	+ return p1;
	+ }
	+
	+ auto Client::syncForever() const -> void
	+ {
	+ using namespace CursorOp;
	+
	+ bool isInitialSync = ! (+m_client[&ClientModel::syncToken]).has_value();
	+
	+ auto syncRes = m_ctx.dispatch(SyncAction{});
	+
	+ auto uploadOneTimeKeysRes = syncRes
	+ .then([*this](auto stat) {
	+ if (! stat.success()) {
	+ return m_ctx.createResolvedPromise(stat);
	+ }
	+ bool hasCrypto{+m_client[&ClientModel::crypto]};
	+ auto p1 = hasCrypto
	+ ? m_ctx.dispatch(GenerateAndUploadOneTimeKeysAction{})
	+ : m_ctx.createResolvedPromise(true);
	+ return p1;
	+ });
	+
	+ auto queryKeysRes = syncRes
	+ .then([*this, isInitialSync](auto stat) {
	+ if (! stat.success()) {
	+ return m_ctx.createResolvedPromise(stat);
	+ }
	+ bool hasCrypto{+m_client[&ClientModel::crypto]};
	+ return hasCrypto
	+ ? m_ctx.dispatch(QueryKeysAction{isInitialSync})
	+ : m_ctx.createResolvedPromise(true);
	+ });
	+
	+ m_ctx.promiseInterface()
	+ .all(std::vector<PromiseT>{uploadOneTimeKeysRes, queryKeysRes})
	+ .then([*this](auto stat) {
	+ if (stat.success()) {
	+ syncForever();
	+ }
	+ });
	}
	}
	diff --git a/src/client/client.hpp b/src/client/client.hpp
	index e547147..08812c9 100644
	--- a/src/client/client.hpp
	+++ b/src/client/client.hpp
	@@ -1,269 +1,279 @@
	/*
	* Copyright (C) 2020 Tusooa Zhu
	*
	* This file is part of libkazv.
	*
	* libkazv is free software: you can redistribute it and/or modify
	* it under the terms of the GNU Affero General Public License as
	* published by the Free Software Foundation, either version 3 of the
	* License, or (at your option) any later version.
	*
	* libkazv is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	* GNU Affero General Public License for more details.
	*
	* You should have received a copy of the GNU Affero General Public License
	* along with libkazv. If not, see <https://www.gnu.org/licenses/>.
	*/

	#pragma once
	#include <libkazv-config.hpp>

	#include <lager/reader.hpp>
	#include <immer/box.hpp>
	#include <immer/map.hpp>
	#include <immer/flex_vector.hpp>
	#include <immer/flex_vector_transient.hpp>

	#include "sdk-model.hpp"
	#include "client/client-model.hpp"
	#include "client/actions/content.hpp"

	#include "room/room.hpp"

	namespace Kazv
	{
	/**
	* Represent a Matrix client.
	*/
	class Client
	{
	public:
	using PromiseT = SingleTypePromise<DefaultRetType>;
	/**
	* Constructor.
	*
	* Construct the client.
	*/
	Client(lager::reader<SdkModel> sdk,
	Context<ClientAction> ctx);

	/* lager::reader<immer::map<std::string, Room>> */
	inline auto rooms() const {
	return m_client
	[&ClientModel::roomList]
	[&RoomListModel::rooms];
	}

	/* lager::reader<RangeT<std::string>> */
	inline auto roomIds() const {
	return rooms().xform(
	zug::map([](auto m) {
	return intoImmer(
	immer::flex_vector<std::string>{},
	zug::map([](auto val) { return val.first; }),
	m);
	}));
	}

	KAZV_WRAP_ATTR(ClientModel, m_client, serverUrl)
	KAZV_WRAP_ATTR(ClientModel, m_client, loggedIn)
	KAZV_WRAP_ATTR(ClientModel, m_client, userId)
	KAZV_WRAP_ATTR(ClientModel, m_client, token)
	KAZV_WRAP_ATTR(ClientModel, m_client, deviceId)
	KAZV_WRAP_ATTR(ClientModel, m_client, toDevice)

	/**
	* Get the room with @c id .
	*
	* This is equivalent to `roomByCursor(lager::make_constant(id))`.
	*
	* @param id The room id.
	* @return A Room representing the room with `id`.
	*/
	Room room(std::string id) const;

	/**
	* Get the room with `id`.
	*
	* The Room returned will change as the content in `id` changes.
	*
	* For example, you can have the Room that is always the first
	* alphabetically in all rooms by:
	*
	* \code{.cpp}
	* auto someProcessing =
	* zug::map([=](auto ids) {
	* std::sort(ids.begin(), ids.end(), [=](auto id1, auto id2) {
	* using namespace Kazv::CursorOp;
	* return (+client.room(id1).name()) < (+client.room(id2).name());
	* });
	* return ids;
	* });
	* auto room =
	* client.roomByCursor(
	* client.roomIds().xform(someProcessing)[0]);
	* \endcode
	*
	* @param id A lager::reader<std::string> containing the room id.
	* @return A Room representing the room with `id`.
	*/
	Room roomByCursor(lager::reader<std::string> id) const;

	/**
	* Login using the password.
	*
	* This will create a new session on the homeserver.
	*
	+ * If the returned Promise resolves successfully, this will
	+ * call `startSyncing()`.
	+ *
	* @param homeserver The base url of the homeserver. E.g. `https://tusooa.xyz`.
	* @param username The username. This can be the full user id or
	* just the local part. E.g. `tusooa`, `@tusooa:tusooa.xyz`.
	* @param password The password.
	* @param deviceName Optionally, a custom device name. If empty, `libkazv`
	* will be used.
	* @return A Promise that resolves when logging in successfully, or
	* when there is an error.
	*/
	PromiseT passwordLogin(std::string homeserver, std::string username,
	std::string password, std::string deviceName) const;

	/**
	* Login using `token` and `deviceId`.
	*
	* This will not make a request. Library users should make sure
	* the information is correct and the token and the device id are valid.
	*
	+ * If the returned Promise resolves successfully, this will
	+ * call `startSyncing()`.
	+ *
	* @param homeserver The base url of the homeserver. E.g. `https://tusooa.xyz`.
	* @param username The full user id. E.g. `@tusooa:tusooa.xyz`.
	* @param token The access token.
	* @param deviceId The device id that is paired with `token`.
	* @return A Promise that resolves when the account information is filled in.
	*/
	PromiseT tokenLogin(std::string homeserver, std::string username,
	std::string token, std::string deviceId) const;

	/**
	* Create a room.
	*
	* @param v The visibility of the room.
	* @param name The name of the room.
	* @param alias The alias of the room.
	* @param invite User ids to invite to this room.
	* @param isDirect Whether this room is a direct chat.
	* @param allowFederate Whether to allow users from other homeservers
	* to join this room.
	* @param topic The topic of the room.
	* @param powerLevelContentOverride The content of the m.room.power_levels
	* state event to override the default.
	* @return A Promise that resolves when the room is created,
	* or when there is an error.
	*/
	PromiseT createRoom(RoomVisibility v,
	std::optional<std::string> name = {},
	std::optional<std::string> alias = {},
	immer::array<std::string> invite = {},
	std::optional<bool> isDirect = {},
	bool allowFederate = true,
	std::optional<std::string> topic = {},
	JsonWrap powerLevelContentOverride = json::object()) const;

	/**
	* Join a room by its id.
	*
	* @param roomId The id of the room to join.
	* @return A Promise that resolves when the room is joined,
	* or when there is an error.
	*/
	PromiseT joinRoomById(std::string roomId) const;

	/**
	* Join a room by its id or alias.
	*
	* @param roomId The id or alias of the room to join.
	* @param serverName A list of servers to use when joining the room.
	* This corresponds to the `via` parameter in a matrix.to url.
	* @return A Promise that resolves when the room is joined,
	* or when there is an error.
	*/
	PromiseT joinRoom(std::string roomId, immer::array<std::string> serverName) const;

	/**
	* Upload content to the content repository.
	*
	* @param content The content to upload.
	* @param uploadId
	* @param filename The name of the file.
	* @param contentType The content type of the file.
	* @return A Promise that resolves when the upload is successful,
	* or when there is an error. If it successfully resolves to `r`,
	* `r.dataStr("mxcUri")` will be the MXC URI of the uploaded
	* content.
	*/
	PromiseT uploadContent(immer::box<Bytes> content,
	std::string uploadId,
	std::optional<std::string> filename = std::nullopt,
	std::optional<std::string> contentType = std::nullopt) const;

	/**
	* Convert a MXC URI to an HTTP(s) URI.
	*
	* The converted URI will be using the homeserver of
	* this Client.
	*
	* @param mxcUri The MXC URI to convert.
	* @return The HTTP(s) URI that has the content indicated
	* by `mxcUri`.
	*/
	inline std::string mxcUriToHttp(std::string mxcUri) const {
	using namespace CursorOp;
	auto [serverName, mediaId] = mxcUriToMediaDesc(mxcUri);
	return (+m_client)
	.job<GetContentJob>()
	.make(serverName, mediaId).url();
	}

	/**
	* Download content from the content repository
	*
	* @param mxcUri The MXC URI of the content.
	* @return A Promise that is resolved after the content
	* is downloaded, or when there is an error.
	*/
	PromiseT downloadContent(std::string mxcUri) const;

	/**
	* Download a thumbnail from the content repository
	*
	* @param mxcUri The MXC URI of the content.
	* @param width,height The dimension wanted for the thumbnail
	* @param method The method to generate the thumbnail. Either `Crop`
	* or `Scale`.
	* @return A Promise that is resolved after the thumbnail
	* is downloaded, or when there is an error.
	*/
	PromiseT downloadThumbnail(std::string mxcUri,
	int width,
	int height,
	std::optional<ThumbnailResizingMethod> method = std::nullopt) const;

	// lager::reader<bool>
	inline auto syncing() const {
	return m_client[&ClientModel::syncing];
	}

	/**
	* Start syncing if the Client is not syncing.
	*
	+ * Syncing will continue indefinitely until there is an error.
	+ *
	* @return A Promise that resolves when the Client is syncing
	* (more exactly, when syncing() contains true), or when there is an error.
	*/
	PromiseT startSyncing() const;

	private:
	+ void syncForever() const;
	+
	lager::reader<SdkModel> m_sdk;
	lager::reader<ClientModel> m_client;
	Context<ClientAction> m_ctx;
	};

	}
	diff --git a/src/client/sdk-model.cpp b/src/client/sdk-model.cpp
	index 0540835..9e9910e 100644
	--- a/src/client/sdk-model.cpp
	+++ b/src/client/sdk-model.cpp
	@@ -1,120 +1,87 @@
	/*
	* Copyright (C) 2020 Tusooa Zhu <tusooa@vista.aero>
	*
	* This file is part of libkazv.
	*
	* libkazv is free software: you can redistribute it and/or modify
	* it under the terms of the GNU Affero General Public License as
	* published by the Free Software Foundation, either version 3 of the
	* License, or (at your option) any later version.
	*
	* libkazv is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	* GNU Affero General Public License for more details.
	*
	* You should have received a copy of the GNU Affero General Public License
	* along with libkazv. If not, see <https://www.gnu.org/licenses/>.
	*/

	#include <libkazv-config.hpp>

	#include <zug/into_vector.hpp>

	#include "sdk-model.hpp"

	#include <debug.hpp>

	#include "actions/encryption.hpp"
	#include "actions/states.hpp"

	namespace Kazv
	{

	static const int syncInterval = 2000; // ms

	SdkResult SdkModel::update(SdkModel s, SdkAction a)
	{
	return lager::match(a)(
	[&](ClientAction a) -> SdkResult {
	auto [newClient, clientEff] =
	ClientModel::update(std::move(s.client), a);
	s.client = std::move(newClient);

	bool hasCrypto{s.client.crypto};

	auto jobs = s.client.popAllJobs();
	auto triggers = s.client.popAllTriggers();

	auto eff =
	[=](auto &&ctx) {
	auto sdkEffect = SdkEffect(clientEff);
	sdkEffect(ctx);

	auto &jh = getJobHandler(ctx);
	auto &ee = getEventEmitter(ctx);

	auto ph = ctx.promiseInterface();

	auto promises = zug::into_vector(
	zug::map([=, &jh](auto j) {
	return ctx.createWaitingPromise(
	[=, &jh](auto resolve) {
	jh.submit(
	j,
	[=](Response r) {
	resolve(ctx.dispatch(ProcessResponseAction{r}));
	});
	});
	}), jobs);

	auto combinedPromise = ph.all(promises);

	for (auto t : triggers) {
	ee.emit(t);
	- // start a sync (from posting filters) immediately after login successful
	- // apparently, the sync is guaranteed to be processed
	- // after this action has been processed
	- if (std::holds_alternative<LoginSuccessful>(t)) {
	- ctx.dispatch(PostInitialFiltersAction{});
	- }
	- // start a sync `syncInterval` ms after another sync
	- // if we use encryption, also upload one-time keys needed
	- else if (std::holds_alternative<SyncSuccessful>(t)) {
	- auto syncSuccessful = std::get<SyncSuccessful>(t);
	- if (hasCrypto) {
	- ctx.dispatch(GenerateAndUploadOneTimeKeysAction{});
	- }
	- jh.setTimeout(
	- [=]() {
	- ctx.dispatch(SyncAction{});
	- }, syncInterval);
	- }
	- // start a sync or publish identity keys after posting initial filters
	- else if (std::holds_alternative<PostInitialFiltersSuccessful>(t)) {
	- if (hasCrypto) {
	- ctx.dispatch(UploadIdentityKeysAction{});
	- } else {
	- ctx.dispatch(SyncAction{});
	- }
	- }
	- // start sync after publishing identity keys
	- else if (std::holds_alternative<UploadIdentityKeysSuccessful>(t)) {
	- ctx.dispatch(SyncAction{});
	- }
	- else if (std::holds_alternative<ClaimKeysSuccessful>(t)) {
	+ if (std::holds_alternative<ClaimKeysSuccessful>(t)) {
	auto [event, devicesToSend] = std::get<ClaimKeysSuccessful>(t);
	ctx.dispatch(SendToDeviceMessageAction{event, devicesToSend});
	}
	- else if (std::holds_alternative<ShouldQueryKeys>(t) && hasCrypto) {
	- ctx.dispatch(QueryKeysAction{std::get<ShouldQueryKeys>(t).isInitialSync});
	- }
	}
	return combinedPromise;
	};

	return { std::move(s), eff };
	}
	);
	}
	}

File Metadata

Mime Type: text/x-diff
Expires: Sat, Sep 19, 4:47 AM (1 d, 6 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1768875
Default Alt Text: (47 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions