path: root/src/audio-impl-pw.cpp

#include "audio-impl-pw.hpp"
#include "audio-types.hpp"
#include "game-settings.hpp"
#include "pipewire/keys.h"
#include "pipewire/thread-loop.h"
#include "pk.h"

// see header file
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wpedantic"
#include "spa/utils/json.h"
#pragma GCC diagnostic pop

#define PKE_AUDIO_PW_NAME_RESERVE_LEN 256

pke_audio_implementation_pipewire pke_audio_pw{};

int metadata_event_property(void *data, uint32_t subject, const char *key, const char *type, const char *value);

static const struct pw_stream_events stream_events = {
	.version       = PW_VERSION_STREAM_EVENTS,
	.destroy       = nullptr,
	.state_changed = nullptr,
	.control_info  = nullptr,
	.io_changed    = nullptr,
	.param_changed = on_pipewire_stream_param_changed,
	.add_buffer    = nullptr,
	.remove_buffer = nullptr,
	.process       = on_pipewire_process,
	.drained       = nullptr,
	.command       = nullptr,
	.trigger_done  = nullptr,
};
static const struct pw_registry_events registry_events = {
	.version = PW_VERSION_REGISTRY_EVENTS,
	.global = on_registry_event_global,
	.global_remove = on_registry_event_global_removal,
};
static const struct pw_metadata_events metadata_events {
	.version  = PW_VERSION_METADATA_EVENTS,
	.property = metadata_event_property,
};

bool pw_objects_find_pw_object_by_id(void *removed_id, void *item)
{
	return reinterpret_cast<pke_audio_pw_object *>(item)->id == *(uint32_t *)removed_id;
}

void pke_audio_pw_init() {
	int i;

	pke_audio_pw.bkt = pk_mem_bucket_create("pipewire bucket", PK_MEM_DEFAULT_BUCKET_SIZE, PK_MEMBUCKET_FLAG_NONE);
	pke_audio_pw.loop = nullptr;
	pke_audio_pw.pod_buffer = nullptr;
	pke_audio_pw.stream = nullptr;
	pke_audio_pw.core = nullptr;
	pke_audio_pw.registry = nullptr;
	pke_audio_pw.registry_listener = {};
	pke_audio_pw.metadata = nullptr;
	pke_audio_pw.metadata_listener = {};
	pke_audio_pw.pw_objects = {};
	pke_audio_pw.pw_objects.bkt = pke_audio_pw.bkt;
	pke_audio_pw.created_objects = {};
	pke_audio_pw.created_objects.bkt = pke_audio_pw.bkt;
	pke_audio_pw.default_sink_name = pk_new_arr<char>(PKE_AUDIO_PW_NAME_RESERVE_LEN, pke_audio_pw.bkt);
	pke_audio_pw.default_sink_id = 0;
	pke_audio_pw.is_needing_output_remapped = true;

	pke_audio_pw.pod_buffer = pk_new_arr<uint8_t>(pke_audio_mstr.channel_count * PKE_AUDIO_POD_BUFFER_LEN, pke_audio_pw.bkt);

	pw_init(NULL, NULL);

	pke_audio_pw.loop = pw_thread_loop_new("pke-audio-thrd", NULL);
	if (pke_audio_pw.loop == NULL) {
		fprintf(stderr, "[" __FILE__ "][pke_audio_pw_init] failed to create pw_loop");
		return;
	}

	pw_properties *props = pw_properties_new(
		PW_KEY_MEDIA_NAME, "pke-audio-pw",
		PW_KEY_APP_NAME, "pke",
		PW_KEY_MEDIA_TYPE, "Audio",
		PW_KEY_MEDIA_CATEGORY, "Playback",
		PW_KEY_MEDIA_ROLE, "Music",
		nullptr);
	assert(props != NULL);
	pke_audio_pw.stream = pw_stream_new_simple(
		pw_thread_loop_get_loop(pke_audio_pw.loop),
		"pke-audio-pw",
		props,
		&stream_events,
		NULL /* user-data */);
	if (pke_audio_pw.stream == NULL) {
		fprintf(stderr, "[" __FILE__ "][pke_audio_pw_init] failed to create pw_stream");
		return;
	}

	spa_pod_builder b{};
	b.data = pke_audio_pw.pod_buffer;
	b.size = PKE_AUDIO_POD_BUFFER_LEN * pke_audio_mstr.channel_count;
	b._padding = 0;
	b.callbacks.data = NULL;
	b.callbacks.funcs = NULL;
	b.state.flags = 0;
	b.state.offset = 0;
	b.state.frame = NULL;

	spa_audio_info_raw spa_audio_info_raw_init{};
	spa_audio_info_raw_init.flags = 0;
	spa_audio_info_raw_init.format = SPA_AUDIO_FORMAT_F32;
	spa_audio_info_raw_init.rate = PKE_AUDIO_BITRATE;
	spa_audio_info_raw_init.channels = pke_audio_mstr.channel_count;
	spa_audio_info_raw_init.position[0] = SPA_AUDIO_CHANNEL_FL;
	spa_audio_info_raw_init.position[1] = SPA_AUDIO_CHANNEL_FR;

	const struct spa_pod *params[1];
	params[0] = spa_format_audio_raw_build(&b, SPA_PARAM_EnumFormat,
		&spa_audio_info_raw_init);

	if (i = pw_stream_connect(pke_audio_pw.stream,
		PW_DIRECTION_OUTPUT,
		PW_ID_ANY,
		pw_stream_flags(PW_STREAM_FLAG_MAP_BUFFERS | PW_STREAM_FLAG_RT_PROCESS),
		params, 1), i != 0) {
		fprintf(stderr, "[" __FILE__ "][pke_audio_pw_init] failed to connect to pw_stream");
		return;
	}

	pke_audio_pw.core = pw_stream_get_core(pke_audio_pw.stream);
	if (pke_audio_pw.core == NULL) {
		fprintf(stderr, "[" __FILE__ "][pke_audio_pw_init] failed to get core");
		return;
	}

	pke_audio_pw.registry = pw_core_get_registry(pke_audio_pw.core, PW_VERSION_REGISTRY, 0);
	if (pke_audio_pw.registry == NULL) {
		fprintf(stderr, "[" __FILE__ "][pke_audio_pw_init] failed to get registry");
		return;
	}

	memset(&pke_audio_pw.registry_listener, 0, sizeof(pke_audio_pw.registry_listener));
	pw_registry_add_listener(pke_audio_pw.registry, &pke_audio_pw.registry_listener, &registry_events, NULL);

	pw_thread_loop_start(pke_audio_pw.loop);
	return;
}

void pke_audio_pw_teardown() {
	uint32_t i;
	pke_audio_mstr.mtx_buffer.lock();
	pw_thread_loop_stop(pke_audio_pw.loop);
	for (i = pke_audio_pw.created_objects.next; i > 0; --i) {
		pw_core_destroy(pke_audio_pw.core, ((void **)pke_audio_pw.created_objects.data)[i - 1]);
	}
	for (i = pke_audio_pw.pw_objects.next; i > 0; --i) {
		pw_properties_free(pke_audio_pw.pw_objects[i - 1].props);
	}
	if (pke_audio_pw.metadata != NULL) {
		pw_proxy_destroy((struct pw_proxy*)pke_audio_pw.metadata);
	}
	if (pke_audio_pw.registry != NULL) {
		pw_proxy_destroy((struct pw_proxy*)pke_audio_pw.registry);
	}
	if (pke_audio_pw.core != NULL) {
		pw_core_disconnect(pke_audio_pw.core);
	}
	if (pke_audio_pw.stream != NULL) {
		pw_stream_destroy(pke_audio_pw.stream);
	}
	if (pke_audio_pw.loop != NULL) {
		pw_thread_loop_destroy(pke_audio_pw.loop);
	}
	pw_deinit();
	pk_delete_arr<char>(pke_audio_pw.default_sink_name, PKE_AUDIO_PW_NAME_RESERVE_LEN, pke_audio_pw.bkt);
	pk_delete_arr<uint8_t>(pke_audio_pw.pod_buffer, pke_audio_mstr.channel_count * PKE_AUDIO_POD_BUFFER_LEN, pke_audio_pw.bkt);
	pk_arr_reset(&pke_audio_pw.created_objects);
	pke_audio_pw.created_objects.bkt = nullptr;
	pk_arr_reset(&pke_audio_pw.pw_objects);
	pke_audio_pw.pw_objects.bkt = nullptr;
	pk_mem_bucket_destroy(pke_audio_pw.bkt);
	pke_audio_pw.bkt = CAFE_BABE(pk_membucket);
	pke_audio_mstr.mtx_buffer.unlock();
}

void pke_audio_pw_remap_outputs() {
	uint32_t stream_id;
	uint32_t i;
	uint32_t port_count = {0};
	void *created_obj = {0};
	if (pke_audio_pw.pw_objects.data == nullptr) return;
	if (pke_audio_pw.default_sink_name == nullptr || pke_audio_pw.default_sink_name[0] == '\0') return;
	pw_properties *props = pw_properties_new(NULL, NULL);
	pke_audio_pw_object *objs = (pke_audio_pw_object *)pke_audio_pw.pw_objects.data;
	const char *str;
	uint32_t *out_ports = pk_new_arr<uint32_t>(pke_audio_mstr.channel_count, pkeSettings.mem_bkt.game_transient);
	uint32_t *in_ports = pk_new_arr<uint32_t>(pke_audio_mstr.channel_count, pkeSettings.mem_bkt.game_transient);
	for (i = 0; i < pke_audio_mstr.channel_count; ++i) {
		out_ports[i] = 0;
		in_ports[i] = 0;
	}

	pke_audio_mstr.mtx_buffer.lock();
	pw_thread_loop_lock(pke_audio_pw.loop);

	// remove all existing links
	{
		for (i = pke_audio_pw.created_objects.next; i > 0; --i) {
			pw_core_destroy(pke_audio_pw.core, pke_audio_pw.created_objects[i - 1]);
		}
		pk_arr_clear(&pke_audio_pw.created_objects);
	}

	// find node we're going to connect to
	//  + build out_ports and in_ports
	{
		stream_id = pw_stream_get_node_id(pke_audio_pw.stream);
		if (stream_id == 0) {
			goto remap_done;
		}
		for (i = 0; i < pke_audio_pw.pw_objects.next; ++i) {
			if (objs[i].type != PKE_AUDIO_PW_OBJECT_TYPE_NODE) continue;
			if ((str = spa_dict_lookup(&objs[i].props->dict, "node.name")), str == NULL) continue;
			if (strcmp(pke_audio_pw.default_sink_name, str) == 0) {
				pke_audio_pw.default_sink_id = objs[i].id;
				break;
			}
		}
		// assert(stream_id != 0);
		if (pke_audio_pw.default_sink_id == 0) {
			goto remap_done;
		}
		for (i = 0; i < pke_audio_pw.pw_objects.next; ++i) {
			if (objs[i].type != PKE_AUDIO_PW_OBJECT_TYPE_PORT) continue;
			if (objs[i].data.port.node != pke_audio_pw.default_sink_id && objs[i].data.port.node != stream_id) continue;
			if (objs[i].data.port.node == pke_audio_pw.default_sink_id && objs[i].data.port.direction == PW_DIRECTION_OUTPUT) continue;
			if (objs[i].data.port.node == stream_id) {
				out_ports[objs[i].data.port.id] = objs[i].id;
			} else {
				port_count++;
				in_ports[objs[i].data.port.id] = objs[i].id;
			}
		}
	}
	// assert(port_count != 0);
	if (port_count == 0) {
		goto remap_done;
	}

	// do links
	{
		for (i = 0; i < pke_audio_mstr.channel_count; ++i) {
			if (out_ports[i] == 0) {
				goto remap_done;
			}
			fprintf(stdout, "[pke_audio_pw_remap_outputs] Mapping (out:in): %d:%d\n", out_ports[i], in_ports[i]);
			pw_properties_clear(props);
			pw_properties_setf(props, PW_KEY_LINK_OUTPUT_PORT, "%d", out_ports[i]);
			pw_properties_setf(props, PW_KEY_LINK_INPUT_PORT, "%d", in_ports[i % port_count]);
			created_obj = pw_core_create_object(pke_audio_pw.core, "link-factory", PW_TYPE_INTERFACE_Link, PW_VERSION_LINK, &props->dict, 0);
			if (created_obj != NULL) pk_arr_append(&pke_audio_pw.created_objects, created_obj);
		}
	}
	fprintf(stdout, "[pke_audio_pw_remap_outputs] remapped %i objs\n", pke_audio_pw.created_objects.next);

remap_done:
	if (props != nullptr) pw_properties_free(props);
	pke_audio_pw.is_needing_output_remapped = pke_audio_pw.created_objects.next != pke_audio_mstr.channel_count;
	pw_thread_loop_unlock(pke_audio_pw.loop);
	pke_audio_mstr.mtx_buffer.unlock();
}

void on_pipewire_process(void *user_data) {
	(void)user_data;
	pw_buffer *b;
	spa_buffer *buf;
	int stride;
	int64_t n_frames;
	float *dst;

	if ((b = pw_stream_dequeue_buffer(pke_audio_pw.stream)) == NULL) {
		fprintf(stderr, "[" __FILE__ "][on_pipewire_process] out of buffers");
		return;
	}

	buf = b->buffer;
	if ((dst = (float*)buf->datas[0].data) == NULL) {
		fprintf(stderr, "[" __FILE__ "][on_pipewire_process] given buffer was null");
		return;
	}

	pke_audio_mstr.mtx_buffer.lock();

	stride = sizeof(float) * pke_audio_mstr.channel_count;

	n_frames = buf->datas[0].maxsize / stride;
	if (b->requested) {
		n_frames = PK_MIN((int64_t)b->requested, n_frames);
	}
	// fprintf(stdout, "[pw] frame count available: %li, requested: %lu, ready: %li\n", n_frames, b->requested, pke_audio_mstr.buffer_frames);
	n_frames = PK_MIN(n_frames, pke_audio_mstr.buffer_frames);

	buf->datas[0].chunk->offset = 0;
	buf->datas[0].chunk->stride = stride;
	buf->datas[0].chunk->size = n_frames * stride;

	if (buf->datas[0].chunk->size == 0)  {
		goto audio_done;
	}

	memcpy(dst, pke_audio_mstr.buffer, buf->datas[0].chunk->size);

	if (n_frames > 0 && n_frames < pke_audio_mstr.buffer_frames) {
		// TODO PERF I think if I always had two buffers allocated
		//  I could alternate between them for better perf.
		float* new_buffer = pk_new_arr<float>(PKE_AUDIO_BUFFER_FRAMES * pke_audio_mstr.channel_count * sizeof(float), pke_audio_mstr.bkt);
		memcpy(
			new_buffer,
			pke_audio_mstr.buffer + (pke_audio_mstr.channel_count * n_frames),
			stride * (pke_audio_mstr.buffer_frames - n_frames)
		);
		pk_delete_arr<float>(pke_audio_mstr.buffer, PKE_AUDIO_BUFFER_FRAMES * pke_audio_mstr.channel_count * sizeof(float), pke_audio_mstr.bkt);
		pke_audio_mstr.buffer = new_buffer;

		// fprintf(stdout, "[pw] shift buffer. buffer_frames before: %li, processed_frames: %li, after: %li\n", pke_audio_mstr.buffer_frames, n_frames, pke_audio_mstr.buffer_frames - n_frames);
		pke_audio_mstr.buffer_frames -= n_frames;
	} else {
		pke_audio_mstr.buffer_frames = 0;
	}

audio_done:
	pw_stream_queue_buffer(pke_audio_pw.stream, b);
	pke_audio_mstr.mtx_buffer.unlock();
	return;
}

void on_pipewire_stream_param_changed(void *, uint32_t id, const struct spa_pod *param) {
	/* NULL means to clear the format */
	if (param == NULL || id != SPA_PARAM_Format) {
		// memset(&pke_audio_pw.format, 0, sizeof(struct spa_audio_info));
		return;
	}

	if (spa_format_parse(param, &pke_audio_pw.format.media_type, &pke_audio_pw.format.media_subtype) < 0) {
		return;
	}

	/* only accept raw audio */
	if (pke_audio_pw.format.media_type != SPA_MEDIA_TYPE_audio || pke_audio_pw.format.media_subtype != SPA_MEDIA_SUBTYPE_raw) {
		return;
	}

	/* call a helper function to parse the format for us. */
	spa_format_audio_raw_parse(param, &pke_audio_pw.format.info.raw);

	fprintf(stdout, "capturing rate changed:%d channels:%d\n", pke_audio_pw.format.info.raw.rate, pke_audio_pw.format.info.raw.channels);
	return;
}

int metadata_event_property(void *data, uint32_t subject, const char *key, const char *type, const char *value)
{
	if (subject == PW_ID_CORE) {
		if (key == NULL || spa_streq(key, "default.audio.sink")) {
			if (pke_audio_pw.default_sink_id != 0) {
				// the default just changed??
				pke_audio_pw.is_needing_output_remapped = true;
			}
			if (value == NULL || spa_json_str_object_find(value, strlen(value), "name", pke_audio_pw.default_sink_name, PKE_AUDIO_PW_NAME_RESERVE_LEN) < 0) {
				pke_audio_pw.default_sink_name[0] = '\0';
			}
		}
	}
	fprintf(stdout, "PW_METADATA_CB: data: %p, subject: %d, key: '%s', type: '%s', value: '%s'\n", data, subject, key, type, value);
	return 0;
}

void on_registry_event_global(void *data, uint32_t id, uint32_t permissions, const char *type, uint32_t version, const struct spa_dict *props) {
	(void)data;
	(void)permissions;
	const char *str;
	pke_audio_pw_object obj{};
	enum PK_STN_RES res;

	// if (props == NULL) return; // why
	if (props == NULL) {
		printf("PW CALLBACK: NULL PROPS: object: id:%u type:%s/%d\n", id, type, version);
		return; // why - in what scenario does this happen?
	}

	if (spa_streq(type, PW_TYPE_INTERFACE_Node)) {
		printf("PW CALLBACK NODE: object: id:%u type:%s/%d\n", id, type, version);
		obj.type = PKE_AUDIO_PW_OBJECT_TYPE_NODE;
	} else if (spa_streq(type, PW_TYPE_INTERFACE_Port)) {
		obj.type = PKE_AUDIO_PW_OBJECT_TYPE_PORT;
		if ((str = spa_dict_lookup(props, PW_KEY_PORT_DIRECTION)) == NULL) {
			return;
		}
		if (spa_streq(str, "in")) {
			obj.data.port.direction = PW_DIRECTION_INPUT;
		} else if (spa_streq(str, "out")) {
			obj.data.port.direction = PW_DIRECTION_OUTPUT;
		} else {
			return;
		}
		if ((str = spa_dict_lookup(props, PW_KEY_NODE_ID)) == NULL) {
			return;
		}
		if (res = pk_stn(&obj.data.port.node, str, 0), res != PK_STN_RES_SUCCESS) {
			fprintf(stderr, "PW_CALLBACK: failed to parse port node from str: '%s', PK_STN_RES: %i\n", str, res);
			return;
		}
		if ((str = spa_dict_lookup(props, PW_KEY_PORT_ID)) == NULL) {
			return;
		}
		if (res = pk_stn(&obj.data.port.id, str, 0), res != PK_STN_RES_SUCCESS) {
			fprintf(stderr, "PW_CALLBACK: failed to parse port id from str: '%s', PK_STN_RES: %i\n", str, res);
			return;
		}
	} else if (spa_streq(type, PW_TYPE_INTERFACE_Link)) {
		obj.type = PKE_AUDIO_PW_OBJECT_TYPE_LINK;
		if ((str = spa_dict_lookup(props, PW_KEY_LINK_OUTPUT_PORT)) == NULL) {
			return;
		}
		if (res = pk_stn(&obj.data.link.output_port, str, 0), res != PK_STN_RES_SUCCESS) {
			fprintf(stderr, "PW_CALLBACK: failed to parse link output port from str: '%s', PK_STN_RES: %i\n", str, res);
			return;
		}
		if ((str = spa_dict_lookup(props, PW_KEY_LINK_INPUT_PORT)) == NULL) {
			return;
		}
		if (res = pk_stn(&obj.data.link.input_port, str, 0), res != PK_STN_RES_SUCCESS) {
			fprintf(stderr, "PW_CALLBACK: failed to parse link input port from str: '%s', PK_STN_RES: %i\n", str, res);
			return;
		}
	} else if (spa_streq(type, PW_TYPE_INTERFACE_Metadata)) {
		obj.type = PKE_AUDIO_PW_OBJECT_TYPE_METADATA;
		if ((str = spa_dict_lookup(props, PW_KEY_METADATA_NAME)) == NULL) {
			return;
		}
		if (!spa_streq(str, "default")) {
			return;
		}
		pke_audio_pw.metadata = (pw_metadata*)pw_registry_bind(pke_audio_pw.registry, id, PW_TYPE_INTERFACE_Metadata, PW_VERSION_METADATA, 0);

		spa_zero(pke_audio_pw.metadata_listener);

		pw_metadata_add_listener(pke_audio_pw.metadata, &pke_audio_pw.metadata_listener, &metadata_events, NULL);
	} else {
		return;
	}

#if 0
		int i;
		for (i = 0; i < props->n_items; ++i) {
			printf("\t'%s': '%s'", props->items[i].key, props->items[i].value);
		}
		printf("");
#endif

	obj.id = id;
	obj.props = pw_properties_new_dict(props);

	printf("PW CALLBACK: object: id:%u type:%s/v%d\n", id, type, version);

	pk_arr_append_t(&pke_audio_pw.pw_objects, obj);
	// TODO 2025-07-07 JCB
	// I don't know if this logic is ever true.
	// Specifically, on startup we don't have default_sink_id yet.
	// So the only time this catches is if ports on our default node change during runtime?
	// I feel like this was the right idea, but maybe the wrong place for it?
	// The bug I was trying to fix at the time was a race condition where we were
	//  trying to link output ports that didn't exist yet.
	// The correct answer was to validate the data before making the links.
	/*
	if (obj.type == PKE_AUDIO_PW_OBJECT_TYPE_PORT
			&& obj.data.port.node == pke_audio_pw.default_sink_id) {
		pke_audio_pw.is_needing_output_remapped = true;
	}
	*/
	return;
}

void on_registry_event_global_removal(void *data, uint32_t id) {
	(void)data;
	uint32_t i;
	// printf("PW CALLBACK: object remove: id:%u ...", id);
	if (i = pk_arr_find_first_index(&pke_audio_pw.pw_objects, &id, pw_objects_find_pw_object_by_id), i != 0xFFFFFFFF)
	{
		if (pke_audio_pw.pw_objects[i].id == pke_audio_pw.default_sink_id) {
			// we just lost our target audio device
			pke_audio_pw.default_sink_id = 0;
			pke_audio_pw.default_sink_name[0] = '\0';
			pke_audio_pw.is_needing_output_remapped = true;
		}
		pk_arr_remove_at(&pke_audio_pw.pw_objects, i);
		// printf(" removed.\n");
	}
	return;
}