 the correct conversation model ("talk to", not "ask/tell").
Written for the 2009 IntroComp (where it placed first of three), it is by definition an unfinished work. As a result, I gave it an extra star for the potential that I sensed this work could have as a cleaned-up, full-length story.
The true test of an intro is whether you would be interested in continuing, and, in my case, the answer is a definite yes.
// Copyright 2022 Haseeb LLC
// @author: <NAME> <<EMAIL>>

#include "graph_query.h"

int main(int argc, char* argv[]) {
	dummy_tests();

	// Building a query trace where [frontend]-()->[]-()->[emailservice]-()
	trace_structure query_trace;
	query_trace.num_nodes = 3;
	query_trace.node_names.insert(std::make_pair(0, "frontend"));
	query_trace.node_names.insert(std::make_pair(1, ASTERISK_SERVICE));
	// query_trace.node_names.insert(std::make_pair(2, "recommendationservice"));
	query_trace.node_names.insert(std::make_pair(2, "emailservice"));

	query_trace.edges.insert(std::make_pair(0, 1));
	query_trace.edges.insert(std::make_pair(1, 2));

	auto client = gcs::Client();
	std::vector<std::string> total = get_traces_by_structure(query_trace, 1551500618, 1651500700, &client);
	std::cout << "Total results: " << total.size() << std::endl;
	// for (std::string i: total) {
	// 	std::cout << i << std::endl;
	// }
	return 0;
}

/**
 * @brief Get the traces by structure. 
 * 
 * @param query_trace 
 * @param start_time 
 * @param end_time 
 * @param client 
 * @return std::vector<std::string> 
 */
std::vector<std::string> get_traces_by_structure(
	trace_structure query_trace, int start_time, int end_time, gcs::Client* client) {
	std::vector<std::future<std::vector<std::string>>> response_futures;

	for (auto&& object_metadata : client->ListObjects(TRACE_HASHES_BUCKET)) {
		response_futures.push_back(std::async(
			std::launch::async, process_trace_hashes_object_and_retrieve_relevant_trace_ids,
			object_metadata, query_trace, start_time, end_time, client));
	}

	std::vector<std::string> response;
	for_each(response_futures.begin(), response_futures.end(),
		[&response](std::future<std::vector<std::string>>& fut){
			std::vector<std::string> trace_ids = fut.get();
			response.insert(response.end(), trace_ids.begin(), trace_ids.end());
	});
	return response;
}

std::vector<std::string> process_trace_hashes_object_and_retrieve_relevant_trace_ids(
	StatusOr<gcs::ObjectMetadata> object_metadata,
	trace_structure query_trace,
	int start_time,
	int end_time,
	gcs::Client* client
) {
	if (!object_metadata) {
		std::cerr << object_metadata.status().message() << std::endl;
		exit(1);
	}

	std::vector<std::string> response_trace_ids;

	std::string object_name = object_metadata->name();
	std::string batch_name = extract_batch_name(object_name);

	std::pair<int, int> batch_time = extract_batch_timestamps(batch_name);
	if (false == is_object_within_timespan(batch_time, start_time, end_time)) {
		return std::vector<std::string>();
	}

	response_trace_ids = get_trace_ids_from_trace_hashes_object(object_name, client);
	if (response_trace_ids.size() < 1) {
		return response_trace_ids;
	}

	std::string object_content = read_object(TRACE_STRUCT_BUCKET, batch_name, client);

	if (false == does_trace_structure_conform_to_graph_query(object_content, response_trace_ids[0], query_trace, client)) {
		return std::vector<std::string>();
	}

	response_trace_ids = filter_trace_ids_based_on_query_timestamp(
		response_trace_ids, batch_name, object_content, start_time, end_time, client);
	return response_trace_ids;
}

/**
 * TODO: maybe pass a pointer to object_content only, would that be more performant?
 */
bool does_trace_structure_conform_to_graph_query(
	std::string object_content, std::string trace_id, trace_structure query_trace, gcs::Client* client ) {
	std::string trace = extract_trace_from_traces_object(trace_id, object_content);

	trace_structure candidate_trace = morph_trace_object_to_trace_structure(trace);
	bool response = is_isomorphic(query_trace, candidate_trace);
	return response;
}

std::vector<std::string> split_by_line(std::string input) {
	std::vector<std::string> result = split_by_char(input, "\n");
	if (result[result.size()-1].length() < 1) {
		result.pop_back();
	}

	return result;
}

/**
 * @brief Right now this function only calculates whether the batch timestamps have some overlap with the 
 * timespan provided in the query. 
 * 
 * @return bool
 */
bool is_object_within_timespan(std::pair<int, int> batch_time, int start_time, int end_time) {
	std::pair<int, int> query_timespan = std::make_pair(start_time, end_time);

	// query timespan between object timespan
	if (batch_time.first <= query_timespan.first && batch_time.second >= query_timespan.second) {
		return true;
	}

	// query timespan contains object timespan
	if (batch_time.first >= query_timespan.first && batch_time.second <= query_timespan.second) {
		return true;
	}

	// batch timespan overlaps but starts before query timespan
	if (batch_time.first <= query_timespan.first && batch_time.second <= query_timespan.second
	&& batch_time.second >= query_timespan.first) {
		return true;
	}

	// vice versa
	if (batch_time.first >= query_timespan.first && batch_time.second >= query_timespan.second
	&& batch_time.first <= query_timespan.second) {
		return true;
	}

	return false;
}

std::string read_object(std::string bucket, std::string object, gcs::Client* client) {
	auto reader = client->ReadObject(bucket, object);
	if (!reader) {
		std::cerr << "Error reading object " << bucket << "/" << object << " :" << reader.status() << "\n";
		exit(1);
	}

	std::string object_content{std::istreambuf_iterator<char>{reader}, {}};
	return object_content;
}

std::vector<std::string> get_trace_ids_from_trace_hashes_object(std::string object_name, gcs::Client* client) {
	std::string object_content = read_object(TRACE_HASHES_BUCKET, object_name, client);
	std::vector<std::string> trace_ids = split_by_line(object_content);

	return trace_ids;
}

std::string extract_batch_name(std::string object_name) {
	std::vector<std::string> result;
	boost::split(result, object_name, boost::is_any_of("/"));

	return result[1];
