init

.gitignore

@@ -0,0 +1 @@
+syncplay

CLARIFY.md

@@ -0,0 +1,134 @@
+## Missing specifications that block implementation
+
+1.  Media selection
+
+The client command has no movie argument. It is unclear whether the user:
+
+- passes a file to syncplay;
+- opens a file afterward through mpv;
+- or launches mpv separately.
+
+ANSWER: passes a file to syncplay
+
+2.  Session model
+
+The server broadcasts to “all connected clients,” but it does not say whether there is exactly one global viewing session or whether named rooms are required.
+
+ANSWER: no rooms. one global viewing session. This program is intended to be run by people that know each other, and maximum 2-3 persons.
+
+3. Joining behavior
+
+A newly connected client needs an initial state: current position, paused/playing status, and possibly media identity. A pure event dispatcher cannot provide that unless another client republishes its state.
+
+ANSWER: `syncplay` starts and has a movie specified. The initial state is paused and at 00:00:00.
+
+4. Authoritative state
+
+There is no rule for simultaneous or conflicting actions. For example, A pauses while B seeks or immediately presses play. The server needs an ordering rule, even if it is only “last event received wins.”
+
+ANSWER: even if the movie is paused, users can still seek. This is how normally the video player works. So there is no reason why a user could seek while the player is paused. Agree?
+
+5. Network protocol
+
+The plan does not specify:
+
+- TCP, WebSocket, or another transport;
+- message framing;
+- message schema;
+- protocol versioning;
+- connection handshake; - keepalive or timeout behavior.
+
+ANSWER: TCP preferrable, but pick the easiest option. For all other above points i have no answer, pick sensible defaults.
+
+3.  Feedback-loop prevention
+
+Remotely applying pause or seek causes mpv events locally. The plan does not define how clients recognize and suppress those echoed changes.
+
+ANSWER: i don't understand where is the issue. The event is firts applied locally (i.e. pause). Then that same event is sent to the server for the other clients to pick it up.
+
+4.  Meaning of synchronization
+
+Relaying play, pause, and seek operations does not maintain close synchronization during uninterrupted playback. Independent players can gradually drift, and network latency makes “play now” happen at different times.
+The plan must say whether the prototype only mirrors user actions or also periodically measures and corrects drift.
+
+ANSWER: I don't care about the exact synchronization or drifting. If drifting happens, the users can still seek and all other clients will receive the event.
+
+5.  Seek semantics
+
+“Seek to xx:xx:xx” suggests absolute seconds, but the protocol needs to specify:
+
+- absolute versus relative positions;
+- precision;
+- exact versus keyframe seek;
+- behavior when duration differs between clients.
+
+ANSWER: OK.
+
+6.  Source compatibility
+
+There is no requirement that participants load the same media. A position alone is meaningless if files, editions, durations, or opening offsets differ.
+
+ANSWER: it is assumed that the clients pick the same file.
+
+7.  Reconnect behavior
+
+It is unspecified whether a disconnected client retries, exits, keeps playing locally, or resynchronizes after reconnecting.
+
+ANSWER: if a client disconnects, or loses the connection to the server, the playback continues normally. Log the event to the console.
+
+8.  mpv startup lifecycle
+
+The plan does not define:
+
+- how long to wait for the IPC socket;
+- what happens if mpv fails to start;
+- how stale socket files are handled;
+- whether closing mpv terminates syncplay;
+- whether closing syncplay terminates mpv.
+
+ANSWERS:
+
+- how long to wait for the IPC socket: I don't understand the question. Use sensible defaults.
+- what happens if mpv fails to start: Quit and log the error.
+- how stale socket files are handled: Quit and log the error.
+- whether closing mpv terminates syncplay: Yes.
+- whether closing syncplay terminates mpv: Yes.
+
+9.  Platform scope
+
+/Applications/mpv.app/... is macOS-specific, while Unix sockets suggest macOS/Linux. Supported client platforms and executable discovery need to be explicit.
+
+ANSWER: Support only MacOS and Linux (Linux only for server) with the defaults i have provided.
+
+10. Configuration
+
+Hard-coding /tmp/mpv-socket prevents multiple local instances and risks collision with a stale or unrelated socket. The mpv executable and socket should at least be overrideable, even if defaults are provided.
+
+ANSWER: I don't care about multiple local instances. Only one is expected. Provide `--socket` and `--mpv` cli options to override defaults.
+
+11. Security boundary
+
+A server bound to 0.0.0.0 permits anyone who can reach it to control every connected player unless authentication or network-level trust is assumed. The plan needs to state whether an unauthenticated trusted-network prototype is acceptable.
+
+ANSWER: I don't care about security or authentication. This is expected to run on a trusted environment by a few people.
+
+12. Failure handling and limits
+
+Missing behavior includes malformed messages, slow clients, disconnected sockets, oversized messages, and server capacity.
+
+ANSWER: TCP or WebSocket should deal with malformed messages already. IDC about slow clients, we are talking about very small packets. Oversized messages and server capacity are not a concern for all answers above.
+
+13. Prototype acceptance criteria
+
+There is no concrete test defining “up and running”: supported OS, number of clients, acceptable latency/drift, reconnect expectations, and required automated tests.
+
+ANSWER: supported OS: MacOS and Linux (Linux server only). I said already IDC about drifting. Write automated unit tests in Odin where applicable. I don't need end-to-end testing.
+
+## Internal inconsistencies or misleading details
+
+- The IPC socket path is supplied to mpv, but mpv creates/listens on it; syncplay does not create a shared read/write command stream. ANSWER IDK, pick defaults.
+- “Every interaction with MPV will write in the socket” is incorrect; explicit property observation and event handling are needed. ANSWER IDK, pick defaults.
+- The example contains --input-terminal=no twice. ANSWER: remove the duplicate option, it is a typo.
+- “The socket path and mpv path must likely be constants” conflicts with portability and safe concurrent use. ANSWER: we provide overrides via cli options.
+- Calling the server only an “events dispatcher” conflicts with seamless late joining, which requires either server-held state or a designated client supplying a snapshot. ANSWER: IDC about late joining. If someone joins late, a seek event from any client will fix the drift.
+- “Synchronized playback” implies clock/drift synchronization, while the listed behavior only promises event replication. ANSWER IDC about clock/drift sync.

PLAN.md

@@ -0,0 +1,83 @@
+# Sync play for MPV
+
+Goal: write an Odin application to sync playback between remote viewers via a small Odin program.
+
+The program will be called `syncplay`. The player is called `mpv`. `syncplay` is a single executable that can be run by clients, and is also executable in a remote server that functions as events dispatcher. The server is required.
+
+Use case: two friends want to see the same movie using MPV, and they want synchronized playback. If one pauses, the video pauses for the other viewer. If one seeks to xx:xx:xx, also the other player seeks to xx:xx:xx.
+
+**Server**
+
+```sh
+./syncplay server --port <port> --host 0.0.0.0
+```
+
+**Client**
+
+```sh
+./syncplay client --port <port> --host <remote-ip> \
+  [--mpv <mpv-path>] [--socket <socket-path>] <movie>
+```
+
+Supported operations:
+
+- play
+- pause
+- seek
+
+On macOS, the client launches MPV with a JSON IPC UNIX socket. `syncplay`
+connects to that socket, observes pause changes and seek events, and sends JSON
+commands to apply events received from the relay server.
+
+The TCP protocol is newline-delimited JSON. There is one global session, no
+authentication, and no server-held playback state. Messages are absolute state
+changes:
+
+```json
+{"version":1,"type":"welcome","client_id":1}
+{"version":1,"type":"pause","paused":true}
+{"version":1,"type":"pause","paused":false}
+{"version":1,"type":"seek","position":123.456}
+```
+
+The server assigns a numeric ID when a client connects and includes the
+triggering ID in every relayed event. Both server and client terminals log each
+play, pause, and seek action with that client ID.
+
+Clients start paused at `00:00:00` and are assumed to use the same movie. There
+is no clock synchronization or drift correction. If a connection is lost,
+playback continues locally and the event is logged; clients do not reconnect.
+
+**Example flow**:
+
+0. Someone starts `syncplay` on a remote server.
+1. User A starts `syncplay`. The program starts and runs the `mpv` player.
+2. Then it connects to the remote `syncplay` instance.
+3. `syncplay` observes pause and seek events reported by MPV over JSON IPC.
+4. `syncplay` translates supported local events and sends them to the server.
+5. The server will stream to all connected clients the same event.
+6. Any connected client receives the event and sends the corresponding JSON IPC
+   command to its local MPV instance.
+
+Remote seeks use a serialized, acknowledged transaction: disable delivery of
+the `seek` event for the IPC connection, apply an `absolute+exact` seek, then
+re-enable the event even when the seek fails. This prevents remote seeks from
+being reported as new local seeks and looping through the server.
+
+**Example cli command to run MPV**
+
+Pay attention to the socket path. That, along with the MPV path must likely be constants in the Odin program.
+
+```sh
+/Applications/mpv.app/Contents/MacOS/mpv \
+  --force-window=yes \
+  --idle=yes \
+  --pause=yes \
+  --start=0 \
+  --hr-seek=always \
+  --keep-open=always \
+  --keep-open-pause=yes \
+  --input-ipc-server=/tmp/mpv-socket \
+  --input-terminal=no \
+  --terminal=no
+```

README.md

@@ -0,0 +1,88 @@
+# syncplay
+
+`syncplay` is a small Odin program that relays play, pause, and seek actions
+between a few trusted mpv users. One executable provides both the TCP relay
+server and the macOS client.
+
+## Requirements
+
+- Odin 2026-07 or newer
+- macOS for client mode
+- macOS or Linux for server mode
+- mpv installed on every client
+- The same media file (including the same cut) on every client
+
+## Build and test
+
+```sh
+odin check .
+odin test .
+odin build . -out:syncplay
+```
+
+## Run
+
+Start one relay server:
+
+```sh
+./syncplay server --host 0.0.0.0 --port 9000
+```
+
+Then each viewer starts a client with their local copy of the movie:
+
+```sh
+./syncplay client \
+  --host server.example.test \
+  --port 9000 \
+  /path/to/movie.mkv
+```
+
+The macOS defaults can be overridden:
+
+```sh
+./syncplay client \
+  --host server.example.test \
+  --port 9000 \
+  --mpv /custom/path/to/mpv \
+  --socket /tmp/custom-mpv-socket \
+  /path/to/movie.mkv
+```
+
+The client starts at `00:00:00` in the paused state. Closing mpv closes the
+client. Interrupting the client terminates the mpv process it launched. If the
+server connection is lost after startup, mpv continues normally and the client
+logs that synchronization has stopped.
+
+## Protocol
+
+The trusted-network protocol is newline-delimited JSON over TCP:
+
+```json
+{"version":1,"type":"welcome","client_id":1}
+{"version":1,"type":"pause","paused":true}
+{"version":1,"type":"pause","paused":false}
+{"version":1,"type":"seek","position":3723.125}
+```
+
+Messages are limited to 4096 bytes. The server validates and re-encodes every
+message, preserves processing order, and broadcasts it to every connection
+except its sender. It assigns each connection a numeric client ID and adds the
+triggering ID to relayed events. Server and client terminals log play, pause,
+and seek actions with that ID; the triggering client is marked with `(you)` in
+its own terminal.
+
+Remote seeks are applied using an acknowledged mpv IPC transaction: seek event
+delivery is disabled for this IPC connection, an exact absolute seek is sent,
+and event delivery is restored even if the seek fails. This prevents a remote
+seek from being mistaken for a new local seek and sent back indefinitely.
+
+## Prototype limitations
+
+- One global session; no rooms or late-join state snapshot
+- No authentication, encryption, or Internet-facing security
+- No automatic reconnection
+- No media identity checking
+- No clock synchronization or drift correction
+- mpv can report internally generated operations as seek events
+- A local seek made during the brief remote-seek suppression transaction may
+  not be relayed because mpv events do not identify their origin

cli.odin

@@ -0,0 +1,122 @@
+package main
+
+import "core:fmt"
+import "core:os"
+import "core:strconv"
+import "core:strings"
+
+DEFAULT_SERVER_HOST :: "0.0.0.0"
+DEFAULT_MPV_PATH    :: "/Applications/mpv.app/Contents/MacOS/mpv"
+DEFAULT_SOCKET_PATH :: "/tmp/mpv-socket"
+
+Command_Kind :: enum {
+	None,
+	Server,
+	Client,
+	Help,
+}
+
+Options :: struct {
+	command:     Command_Kind,
+	host:        string,
+	port:        int,
+	mpv_path:    string,
+	socket_path: string,
+	movie_path:  string,
+}
+
+usage :: proc() {
+	fmt.println("syncplay - synchronize basic mpv playback events")
+	fmt.println("")
+	fmt.println("Usage:")
+	fmt.println("  syncplay server --port <port> [--host 0.0.0.0]")
+	fmt.println("  syncplay client --host <host> --port <port> [--mpv <path>] [--socket <path>] <movie>")
+}
+
+parse_cli :: proc(args: []string) -> (options: Options, error_message: string) {
+	options.mpv_path = DEFAULT_MPV_PATH
+	options.socket_path = DEFAULT_SOCKET_PATH
+
+	if len(args) < 2 {
+		return options, "expected server or client subcommand"
+	}
+	switch args[1] {
+	case "server":
+		options.command = .Server
+		options.host = DEFAULT_SERVER_HOST
+	case "client":
+		options.command = .Client
+	case "help", "--help", "-h":
+		options.command = .Help
+		return options, ""
+	case:
+		return options, fmt.tprintf("unknown subcommand %q", args[1])
+	}
+
+	i := 2
+	for i < len(args) {
+		arg := args[i]
+		if strings.has_prefix(arg, "--") {
+			if i+1 >= len(args) {
+				return options, fmt.tprintf("missing value for %s", arg)
+			}
+			value := args[i+1]
+			switch arg {
+			case "--host":
+				options.host = value
+			case "--port":
+				port, ok := strconv.parse_int(value)
+				if !ok || port < 1 || port > 65535 {
+					return options, "port must be an integer from 1 to 65535"
+				}
+				options.port = port
+			case "--mpv":
+				if options.command != .Client {
+					return options, "--mpv is only valid for client"
+				}
+				options.mpv_path = value
+			case "--socket":
+				if options.command != .Client {
+					return options, "--socket is only valid for client"
+				}
+				options.socket_path = value
+			case:
+				return options, fmt.tprintf("unknown option %s", arg)
+			}
+			i += 2
+			continue
+		}
+
+		if options.command != .Client || options.movie_path != "" {
+			return options, fmt.tprintf("unexpected positional argument %q", arg)
+		}
+		options.movie_path = arg
+		i += 1
+	}
+
+	if options.port == 0 {
+		return options, "--port is required"
+	}
+	if options.command == .Client {
+		if options.host == "" {
+			return options, "--host is required for client"
+		}
+		if options.movie_path == "" {
+			return options, "movie path is required"
+		}
+	}
+	return options, ""
+}
+
+validate_client_paths :: proc(options: Options) -> (error_message: string) {
+	if _, err := os.stat(options.movie_path, context.temp_allocator); err != nil {
+		return fmt.tprintf("movie does not exist: %s", options.movie_path)
+	}
+	if _, err := os.stat(options.mpv_path, context.temp_allocator); err != nil {
+		return fmt.tprintf("mpv does not exist: %s", options.mpv_path)
+	}
+	if _, err := os.stat(options.socket_path, context.temp_allocator); err == nil {
+		return fmt.tprintf("socket path already exists: %s", options.socket_path)
+	}
+	return ""
+}

cli_test.odin

@@ -0,0 +1,30 @@
+package main
+
+import "core:testing"
+
+@(test)
+cli_parses_server :: proc(t: ^testing.T) {
+	args := []string{"syncplay", "server", "--port", "9000"}
+	options, err := parse_cli(args)
+	testing.expect_value(t, err, "")
+	testing.expect_value(t, options.command, Command_Kind.Server)
+	testing.expect_value(t, options.host, DEFAULT_SERVER_HOST)
+	testing.expect_value(t, options.port, 9000)
+}
+
+@(test)
+cli_parses_client :: proc(t: ^testing.T) {
+	args := []string{"syncplay", "client", "--host", "example.test", "--port", "9000", "movie.mkv"}
+	options, err := parse_cli(args)
+	testing.expect_value(t, err, "")
+	testing.expect_value(t, options.command, Command_Kind.Client)
+	testing.expect_value(t, options.host, "example.test")
+	testing.expect_value(t, options.movie_path, "movie.mkv")
+}
+
+@(test)
+cli_rejects_bad_port :: proc(t: ^testing.T) {
+	args := []string{"syncplay", "server", "--port", "70000"}
+	_, err := parse_cli(args)
+	testing.expect(t, err != "")
+}

client.odin

@@ -0,0 +1,448 @@
+package main
+
+import "base:runtime"
+import "core:encoding/json"
+import "core:fmt"
+import "core:net"
+import "core:os"
+import "core:sync"
+import "core:sys/posix"
+import "core:thread"
+import "core:time"
+
+Client_State :: struct {
+	mutex: sync.Mutex,
+	client_id: u64,
+
+	network_socket: net.TCP_Socket,
+	network_active: bool,
+	network_write_mutex: sync.Mutex,
+
+	mpv:     ^Mpv_Connection,
+	process: os.Process,
+	stopping: bool,
+
+	ignore_initial_pause: bool,
+	remote_pause_pending: bool,
+	remote_pause_value:   bool,
+	have_last_pause:      bool,
+	last_pause:           bool,
+	local_seek_pending:   bool,
+}
+
+client_close_network :: proc(client: ^Client_State, reason: string) {
+	sync.mutex_lock(&client.mutex)
+	if !client.network_active {
+		sync.mutex_unlock(&client.mutex)
+		return
+	}
+	client.network_active = false
+	socket := client.network_socket
+	sync.mutex_unlock(&client.mutex)
+
+	_ = net.shutdown(socket, .Both)
+	net.close(socket)
+	fmt.printfln("client: server connection lost (%s); playback will continue", reason)
+}
+
+client_send_event :: proc(client: ^Client_State, event: Playback_Event) -> bool {
+	line := encode_protocol_event(event)
+	defer delete(line)
+
+	sync.mutex_lock(&client.network_write_mutex)
+	defer sync.mutex_unlock(&client.network_write_mutex)
+	sync.mutex_lock(&client.mutex)
+	active := client.network_active
+	socket := client.network_socket
+	client_id := client.client_id
+	sync.mutex_unlock(&client.mutex)
+	log_playback_event("client", client_id, event, is_local = true)
+	if !active {
+		return false
+	}
+
+	written, send_err := net.send_tcp(socket, transmute([]byte)line)
+	if send_err != nil || written != len(line) {
+		client_close_network(client, "send failed")
+		return false
+	}
+	return true
+}
+
+client_position_worker :: proc(data: rawptr) {
+	defer runtime.default_temp_allocator_destroy(auto_cast context.temp_allocator.data)
+	client := cast(^Client_State)data
+	position, ok := mpv_get_time_position(client.mpv)
+	if ok {
+		client_send_event(client, Playback_Event{kind = .Seek, position = position})
+	} else {
+		fmt.eprintln("client: could not read position after local seek")
+	}
+}
+
+client_on_pause_change :: proc(client: ^Client_State, paused: bool) {
+	should_send := false
+	sync.mutex_lock(&client.mutex)
+	if client.ignore_initial_pause {
+		client.ignore_initial_pause = false
+		client.have_last_pause = true
+		client.last_pause = paused
+	} else if client.remote_pause_pending && paused == client.remote_pause_value {
+		client.remote_pause_pending = false
+		client.have_last_pause = true
+		client.last_pause = paused
+	} else {
+		client.remote_pause_pending = false
+		if !client.have_last_pause || paused != client.last_pause {
+			client.have_last_pause = true
+			client.last_pause = paused
+			should_send = true
+		}
+	}
+	sync.mutex_unlock(&client.mutex)
+
+	if should_send {
+		client_send_event(client, Playback_Event{kind = .Pause, paused = paused})
+	}
+}
+
+client_on_mpv_event :: proc(client: ^Client_State, event_name: string, object: json.Object) {
+	switch event_name {
+	case "property-change":
+		name_value, has_name := object["name"]
+		data_value, has_data := object["data"]
+		if !has_name || !has_data {
+			return
+		}
+		name, name_ok := name_value.(json.String)
+		paused, pause_ok := data_value.(json.Boolean)
+		if name_ok && pause_ok && string(name) == "pause" {
+			client_on_pause_change(client, bool(paused))
+		}
+
+	case "seek":
+		sync.mutex_lock(&client.mutex)
+		client.local_seek_pending = true
+		sync.mutex_unlock(&client.mutex)
+
+	case "playback-restart":
+		should_query := false
+		sync.mutex_lock(&client.mutex)
+		if client.local_seek_pending {
+			client.local_seek_pending = false
+			should_query = true
+		}
+		sync.mutex_unlock(&client.mutex)
+		if should_query {
+			_ = thread.create_and_start_with_data(
+				rawptr(client),
+				client_position_worker,
+				init_context = context,
+				self_cleanup = true,
+			)
+		}
+	}
+}
+
+client_on_mpv_disconnect :: proc(client: ^Client_State) {
+	sync.mutex_lock(&client.mutex)
+	already_stopping := client.stopping
+	client.stopping = true
+	sync.mutex_unlock(&client.mutex)
+	if !already_stopping {
+		fmt.eprintln("client: mpv IPC connection closed")
+		_ = os.process_terminate(client.process)
+	}
+}
+
+client_apply_remote_event :: proc(client: ^Client_State, event: Playback_Event) -> bool {
+	switch event.kind {
+	case .Pause:
+		sync.mutex_lock(&client.mutex)
+		client.remote_pause_pending = true
+		client.remote_pause_value = event.paused
+		sync.mutex_unlock(&client.mutex)
+		if !mpv_set_pause(client.mpv, event.paused) {
+			fmt.eprintln("client: failed to apply remote pause event")
+			return false
+		}
+		return true
+	case .Seek:
+		if !mpv_apply_remote_seek(client.mpv, event.position) {
+			fmt.eprintln("client: failed to apply remote seek safely")
+			return false
+		}
+		return true
+	}
+	return false
+}
+
+client_network_reader :: proc(data: rawptr) {
+	defer runtime.default_temp_allocator_destroy(auto_cast context.temp_allocator.data)
+	client := cast(^Client_State)data
+	framer: Line_Framer
+	framer_init(&framer)
+	defer framer_destroy(&framer)
+	buffer: [2048]byte
+	reason := "connection closed"
+
+	for {
+		count, recv_err := net.recv_tcp(client.network_socket, buffer[:])
+		if recv_err != nil {
+			reason = "receive failed"
+			break
+		}
+		if count == 0 {
+			break
+		}
+		frames, frame_err := framer_push(&framer, buffer[:count])
+		if frame_err != .None {
+			destroy_frames(frames)
+			reason = "oversized message"
+			break
+		}
+		valid := true
+		for frame in frames {
+			if len(frame) == 0 {
+				continue
+			}
+			welcome_id, is_welcome, welcome_err := parse_welcome(transmute([]byte)frame)
+			if is_welcome {
+				if welcome_err != .None {
+					fmt.eprintfln("client: invalid welcome message: %s", protocol_error_string(welcome_err))
+					reason = "invalid welcome message"
+					valid = false
+					break
+				}
+				sync.mutex_lock(&client.mutex)
+				client.client_id = welcome_id
+				sync.mutex_unlock(&client.mutex)
+				fmt.printfln("client: assigned client ID %d", welcome_id)
+				continue
+			}
+			_ = welcome_err
+
+			event, protocol_err := parse_protocol_event(transmute([]byte)frame)
+			if protocol_err != .None {
+				fmt.eprintfln("client: server sent invalid message: %s", protocol_error_string(protocol_err))
+				reason = "invalid server message"
+				valid = false
+				break
+			}
+			if event.client_id == 0 {
+				fmt.eprintln("client: server event is missing its triggering client ID")
+				reason = "event missing client ID"
+				valid = false
+				break
+			}
+			if !client_apply_remote_event(client, event) {
+				reason = "could not apply remote event"
+				valid = false
+				break
+			}
+			log_playback_event("client", event.client_id, event)
+		}
+		destroy_frames(frames)
+		if !valid {
+			break
+		}
+	}
+	client_close_network(client, reason)
+}
+
+receive_server_welcome :: proc(socket: net.TCP_Socket) -> (client_id: u64, ok: bool) {
+	buffer: [MAX_FRAME_BYTES+1]byte
+	length := 0
+	for length < len(buffer) {
+		count, recv_err := net.recv_tcp(socket, buffer[length:length+1])
+		if recv_err != nil || count == 0 {
+			fmt.eprintln("client: server disconnected before assigning a client ID")
+			return 0, false
+		}
+		if buffer[length] == '\n' {
+			end := length
+			if end > 0 && buffer[end-1] == '\r' {
+				end -= 1
+			}
+			matched: bool
+			protocol_err: Protocol_Error
+			client_id, matched, protocol_err = parse_welcome(buffer[:end])
+			if !matched || protocol_err != .None {
+				fmt.eprintfln("client: invalid server welcome: %s", protocol_error_string(protocol_err))
+				return 0, false
+			}
+			return client_id, true
+		}
+		length += 1
+	}
+	fmt.eprintln("client: server welcome exceeded the message limit")
+	return 0, false
+}
+
+start_mpv_process :: proc(options: Options) -> (process: os.Process, ok: bool) {
+	command := []string{
+		options.mpv_path,
+		"--force-window=yes",
+		"--idle=yes",
+		"--pause=yes",
+		"--start=0",
+		"--hr-seek=always",
+		"--keep-open=always",
+		"--keep-open-pause=yes",
+		"--input-terminal=no",
+		"--terminal=no",
+		fmt.tprintf("--input-ipc-server=%s", options.socket_path),
+		options.movie_path,
+	}
+	process_err: os.Error
+	process, process_err = os.process_start(os.Process_Desc{
+		command = command,
+		stdout = os.stdout,
+		stderr = os.stderr,
+	})
+	if process_err != nil {
+		fmt.eprintfln("client: could not start mpv: %v", process_err)
+		return {}, false
+	}
+	return process, true
+}
+
+wait_for_mpv_socket :: proc(process: os.Process, path: string) -> (posix_fd: int, ok: bool) {
+	deadline := time.time_add(time.now(), 5*time.Second)
+	for time.time_to_unix_nano(time.now()) < time.time_to_unix_nano(deadline) {
+		fd, connected := connect_unix_socket(path)
+		if connected {
+			return int(fd), true
+		}
+		state, wait_err := os.process_wait(process, timeout = 0)
+		if wait_err == nil && state.exited {
+			fmt.eprintln("client: mpv exited before its IPC socket became ready")
+			return -1, false
+		}
+		time.sleep(50*time.Millisecond)
+	}
+	fmt.eprintln("client: timed out waiting for mpv IPC socket")
+	return -1, false
+}
+
+run_client :: proc(options: Options) -> bool {
+	when ODIN_OS != .Darwin {
+		fmt.eprintln("client: client mode is supported only on macOS")
+		return false
+	}
+
+	network_socket, dial_err := net.dial_tcp_from_hostname_with_port_override(options.host, options.port)
+	if dial_err != nil {
+		fmt.eprintfln("client: could not connect to server: %v", dial_err)
+		return false
+	}
+	client_id, welcome_ok := receive_server_welcome(network_socket)
+	if !welcome_ok {
+		net.close(network_socket)
+		return false
+	}
+	fmt.printfln("client: assigned client ID %d", client_id)
+
+	process, process_ok := start_mpv_process(options)
+	if !process_ok {
+		net.close(network_socket)
+		return false
+	}
+
+	fd_value, socket_ok := wait_for_mpv_socket(process, options.socket_path)
+	if !socket_ok {
+		_ = os.process_terminate(process)
+		_, _ = os.process_wait(process)
+		net.close(network_socket)
+		return false
+	}
+
+	client := new(Client_State)
+	client.client_id = client_id
+	client.network_socket = network_socket
+	client.network_active = true
+	client.process = process
+	client.ignore_initial_pause = true
+	mpv := new(Mpv_Connection)
+	mpv.fd = posix.FD(fd_value)
+	mpv.client = client
+	client.mpv = mpv
+
+	mpv_thread := mpv_start_reader(mpv)
+	if mpv_thread == nil {
+		fmt.eprintln("client: could not start mpv IPC reader")
+		mpv_close(mpv)
+		client_close_network(client, "startup failed")
+		_ = os.process_terminate(process)
+		_, _ = os.process_wait(process)
+		return false
+	}
+	if !mpv_observe_pause(mpv) {
+		fmt.eprintln("client: could not observe mpv pause state")
+		mpv_close(mpv)
+		client_close_network(client, "startup failed")
+		_ = os.process_terminate(process)
+		_, _ = os.process_wait(process)
+		thread.join(mpv_thread)
+		thread.destroy(mpv_thread)
+		return false
+	}
+
+	network_thread := thread.create_and_start_with_data(
+		rawptr(client),
+		client_network_reader,
+		init_context = context,
+		self_cleanup = false,
+	)
+	if network_thread == nil {
+		fmt.eprintln("client: could not start server reader")
+		mpv_close(mpv)
+		client_close_network(client, "startup failed")
+		_ = os.process_terminate(process)
+		_, _ = os.process_wait(process)
+		thread.join(mpv_thread)
+		thread.destroy(mpv_thread)
+		return false
+	}
+
+	fmt.println("client: connected; playback starts paused at 00:00:00")
+	state: os.Process_State
+	wait_err: os.Error
+	for {
+		state, wait_err = os.process_wait(process, timeout = 100*time.Millisecond)
+		if wait_err == nil {
+			break
+		}
+		if should_shutdown() {
+			fmt.println("client: shutdown requested; terminating mpv")
+			_ = os.process_terminate(process)
+			state, wait_err = os.process_wait(process, timeout = 2*time.Second)
+			if wait_err != nil {
+				_ = os.process_kill(process)
+				state, wait_err = os.process_wait(process)
+			}
+			break
+		}
+		if wait_err != os.General_Error.Timeout {
+			break
+		}
+	}
+	sync.mutex_lock(&client.mutex)
+	client.stopping = true
+	sync.mutex_unlock(&client.mutex)
+	client_close_network(client, "mpv exited")
+	mpv_close(mpv)
+	thread.join(network_thread)
+	thread.destroy(network_thread)
+	thread.join(mpv_thread)
+	thread.destroy(mpv_thread)
+	if wait_err != nil {
+		fmt.eprintfln("client: waiting for mpv failed: %v", wait_err)
+		return false
+	}
+	if !state.success {
+		fmt.eprintfln("client: mpv exited with status %d", state.exit_code)
+		return false
+	}
+	return true
+}

client_test.odin

@@ -0,0 +1,23 @@
+package main
+
+import "core:testing"
+
+@(test)
+remote_pause_notification_is_suppressed :: proc(t: ^testing.T) {
+	client := Client_State{
+		remote_pause_pending = true,
+		remote_pause_value = true,
+	}
+	client_on_pause_change(&client, true)
+	testing.expect_value(t, client.remote_pause_pending, false)
+	testing.expect_value(t, client.have_last_pause, true)
+	testing.expect_value(t, client.last_pause, true)
+}
+
+@(test)
+initial_pause_notification_is_baseline_only :: proc(t: ^testing.T) {
+	client := Client_State{ignore_initial_pause = true}
+	client_on_pause_change(&client, true)
+	testing.expect_value(t, client.ignore_initial_pause, false)
+	testing.expect_value(t, client.have_last_pause, true)
+}

framing.odin

@@ -0,0 +1,70 @@
+package main
+
+import "core:strings"
+
+Frame_Error :: enum {
+	None,
+	Too_Large,
+}
+
+Line_Framer :: struct {
+	buffer: [dynamic]byte,
+}
+
+framer_init :: proc(framer: ^Line_Framer, allocator := context.allocator) {
+	framer.buffer = make([dynamic]byte, 0, 256, allocator)
+}
+
+framer_destroy :: proc(framer: ^Line_Framer) {
+	delete(framer.buffer)
+	framer.buffer = nil
+}
+
+destroy_frames :: proc(frames: [dynamic]string) {
+	for frame in frames {
+		delete(frame)
+	}
+	delete(frames)
+}
+
+framer_push :: proc(
+	framer: ^Line_Framer,
+	data: []byte,
+	allocator := context.allocator,
+) -> (frames: [dynamic]string, err: Frame_Error) {
+	frames = make([dynamic]string, 0, allocator)
+	append(&framer.buffer, ..data)
+
+	start := 0
+	for i := 0; i < len(framer.buffer); i += 1 {
+		if framer.buffer[i] != '\n' {
+			if i-start >= MAX_FRAME_BYTES {
+				return frames, .Too_Large
+			}
+			continue
+		}
+
+		end := i
+		if end > start && framer.buffer[end-1] == '\r' {
+			end -= 1
+		}
+		if end-start > MAX_FRAME_BYTES {
+			return frames, .Too_Large
+		}
+		frame := strings.clone(string(framer.buffer[start:end]), allocator)
+		append(&frames, frame)
+		start = i+1
+	}
+
+	if start > 0 {
+		remaining := len(framer.buffer)-start
+		if remaining > 0 {
+			copy(framer.buffer[:remaining], framer.buffer[start:])
+		}
+		resize(&framer.buffer, remaining)
+	}
+	if len(framer.buffer) > MAX_FRAME_BYTES {
+		return frames, .Too_Large
+	}
+	return frames, .None
+}

framing_test.odin

@@ -0,0 +1,37 @@
+package main
+
+import "core:testing"
+
+@(test)
+framer_handles_fragmented_and_coalesced_messages :: proc(t: ^testing.T) {
+	framer: Line_Framer
+	framer_init(&framer)
+	defer framer_destroy(&framer)
+
+	first: string = "one\ntw"
+	frames, err := framer_push(&framer, transmute([]byte)first)
+	defer destroy_frames(frames)
+	testing.expect_value(t, err, Frame_Error.None)
+	testing.expect_value(t, len(frames), 1)
+	testing.expect_value(t, frames[0], "one")
+
+	second: string = "o\r\nthree\n"
+	frames2, err2 := framer_push(&framer, transmute([]byte)second)
+	defer destroy_frames(frames2)
+	testing.expect_value(t, err2, Frame_Error.None)
+	testing.expect_value(t, len(frames2), 2)
+	testing.expect_value(t, frames2[0], "two")
+	testing.expect_value(t, frames2[1], "three")
+}
+
+@(test)
+framer_rejects_oversized_message :: proc(t: ^testing.T) {
+	framer: Line_Framer
+	framer_init(&framer)
+	defer framer_destroy(&framer)
+	data := make([]byte, MAX_FRAME_BYTES+1)
+	defer delete(data)
+	frames, err := framer_push(&framer, data)
+	defer destroy_frames(frames)
+	testing.expect_value(t, err, Frame_Error.Too_Large)
+}

main.odin

@@ -0,0 +1,35 @@
+package main
+
+import "core:fmt"
+import "core:os"
+
+main :: proc() {
+	install_shutdown_handlers()
+	options, cli_error := parse_cli(os.args)
+	if cli_error != "" {
+		fmt.eprintln("syncplay:", cli_error)
+		usage()
+		os.exit(2)
+	}
+	if options.command == .Help {
+		usage()
+		return
+	}
+
+	#partial switch options.command {
+	case .Server:
+		if !run_server(options) {
+			os.exit(1)
+		}
+	case .Client:
+		if path_error := validate_client_paths(options); path_error != "" {
+			fmt.eprintln("syncplay:", path_error)
+			os.exit(1)
+		}
+		if !run_client(options) {
+			os.exit(1)
+		}
+	case:
+		unreachable()
+	}
+}

mpv.odin

@@ -0,0 +1,281 @@
+package main
+
+import "base:runtime"
+import "core:encoding/json"
+import "core:fmt"
+import "core:sync"
+import "core:sys/posix"
+import "core:thread"
+import "core:time"
+
+MPV_COMMAND_TIMEOUT :: 5*time.Second
+
+Mpv_Command_Result :: struct {
+	success:  bool,
+	has_data: bool,
+	data:     f64,
+}
+
+Mpv_Connection :: struct {
+	fd: posix.FD,
+
+	command_mutex: sync.Mutex,
+	response_mutex: sync.Mutex,
+	response_cond:  sync.Cond,
+	next_request_id: i64,
+	pending_id:      i64,
+	pending_done:    bool,
+	pending_result:  Mpv_Command_Result,
+	closed:          bool,
+
+	client: ^Client_State,
+}
+
+connect_unix_socket :: proc(path: string) -> (fd: posix.FD, ok: bool) {
+	if len(path) == 0 {
+		return -1, false
+	}
+	address: posix.sockaddr_un
+	if len(path) >= len(address.sun_path) {
+		return -1, false
+	}
+	address.sun_family = .UNIX
+	when ODIN_OS == .Darwin {
+		address.sun_len = u8(size_of(address))
+	}
+	for byte_value, index in transmute([]byte)path {
+		address.sun_path[index] = auto_cast byte_value
+	}
+	address.sun_path[len(path)] = 0
+
+	fd = posix.socket(.UNIX, .STREAM)
+	if fd < 0 {
+		return -1, false
+	}
+	if posix.connect(fd, (^posix.sockaddr)(&address), posix.socklen_t(size_of(address))) != .OK {
+		_ = posix.close(fd)
+		return -1, false
+	}
+	return fd, true
+}
+
+mpv_send_all :: proc(fd: posix.FD, message: string) -> bool {
+	bytes := transmute([]byte)message
+	written: int = 0
+	for written < len(bytes) {
+		remaining := len(bytes)-written
+		count := posix.send(fd, raw_data(bytes[written:]), auto_cast remaining, {.NOSIGNAL})
+		if count <= 0 {
+			return false
+		}
+		written += int(count)
+	}
+	return true
+}
+
+mpv_command_locked :: proc(mpv: ^Mpv_Connection, command_json: string) -> Mpv_Command_Result {
+	sync.mutex_lock(&mpv.response_mutex)
+	if mpv.closed {
+		sync.mutex_unlock(&mpv.response_mutex)
+		return {}
+	}
+	mpv.next_request_id += 1
+	request_id := mpv.next_request_id
+	mpv.pending_id = request_id
+	mpv.pending_done = false
+	mpv.pending_result = {}
+	sync.mutex_unlock(&mpv.response_mutex)
+
+	message := fmt.aprintf("{{\"command\":%s,\"request_id\":%d}}\n", command_json, request_id)
+	defer delete(message)
+	if !mpv_send_all(mpv.fd, message) {
+		return {}
+	}
+
+	sync.mutex_lock(&mpv.response_mutex)
+	defer sync.mutex_unlock(&mpv.response_mutex)
+	deadline_remaining := MPV_COMMAND_TIMEOUT
+	for !mpv.pending_done && !mpv.closed {
+		start := time.now()
+		if !sync.cond_wait_with_timeout(&mpv.response_cond, &mpv.response_mutex, deadline_remaining) {
+			break
+		}
+		elapsed := time.since(start)
+		if elapsed >= deadline_remaining {
+			break
+		}
+		deadline_remaining -= elapsed
+	}
+	if !mpv.pending_done || mpv.pending_id != request_id {
+		return {}
+	}
+	return mpv.pending_result
+}
+
+mpv_command :: proc(mpv: ^Mpv_Connection, command_json: string) -> Mpv_Command_Result {
+	sync.mutex_lock(&mpv.command_mutex)
+	defer sync.mutex_unlock(&mpv.command_mutex)
+	return mpv_command_locked(mpv, command_json)
+}
+
+mpv_apply_remote_seek :: proc(mpv: ^Mpv_Connection, position: f64) -> bool {
+	sync.mutex_lock(&mpv.command_mutex)
+	defer sync.mutex_unlock(&mpv.command_mutex)
+
+	disabled := mpv_command_locked(mpv, `["disable_event","seek"]`)
+	if !disabled.success {
+		return false
+	}
+	reenabled := false
+	defer if !reenabled {
+		result := mpv_command_locked(mpv, `["enable_event","seek"]`)
+		if !result.success {
+			fmt.eprintln("client: failed to re-enable mpv seek events")
+		}
+	}
+
+	seek_command := fmt.aprintf(`["seek",%.6f,"absolute+exact"]`, position)
+	defer delete(seek_command)
+	seek_result := mpv_command_locked(mpv, seek_command)
+	enable_result := mpv_command_locked(mpv, `["enable_event","seek"]`)
+	reenabled = enable_result.success
+	return seek_result.success && enable_result.success
+}
+
+mpv_set_pause :: proc(mpv: ^Mpv_Connection, paused: bool) -> bool {
+	command := fmt.aprintf(
+		`["set_property","pause",%s]`,
+		"true" if paused else "false",
+	)
+	defer delete(command)
+	return mpv_command(mpv, command).success
+}
+
+mpv_get_time_position :: proc(mpv: ^Mpv_Connection) -> (f64, bool) {
+	result := mpv_command(mpv, `["get_property","time-pos"]`)
+	return result.data, result.success && result.has_data
+}
+
+mpv_observe_pause :: proc(mpv: ^Mpv_Connection) -> bool {
+	return mpv_command(mpv, `["observe_property",1,"pause"]`).success
+}
+
+mpv_parse_response :: proc(mpv: ^Mpv_Connection, object: json.Object) -> bool {
+	id_value, has_id := object["request_id"]
+	if !has_id {
+		return false
+	}
+	id, id_ok := id_value.(json.Integer)
+	if !id_ok {
+		return false
+	}
+	error_value, has_error := object["error"]
+	if !has_error {
+		return false
+	}
+	error_name, error_ok := error_value.(json.String)
+	if !error_ok {
+		return false
+	}
+
+	result := Mpv_Command_Result{success = string(error_name) == "success"}
+	if data_value, has_data := object["data"]; has_data {
+		#partial switch value in data_value {
+		case json.Integer:
+			result.has_data = true
+			result.data = f64(value)
+		case json.Float:
+			result.has_data = true
+			result.data = f64(value)
+		}
+	}
+
+	sync.mutex_lock(&mpv.response_mutex)
+	if i64(id) == mpv.pending_id {
+		mpv.pending_result = result
+		mpv.pending_done = true
+		sync.cond_signal(&mpv.response_cond)
+	}
+	sync.mutex_unlock(&mpv.response_mutex)
+	return true
+}
+
+mpv_handle_line :: proc(mpv: ^Mpv_Connection, line: string) {
+	root: json.Value
+	if json.unmarshal(transmute([]byte)line, &root, spec = .JSON) != nil {
+		fmt.eprintln("client: received malformed JSON from mpv")
+		return
+	}
+	defer json.destroy_value(root)
+	object, ok := root.(json.Object)
+	if !ok {
+		return
+	}
+	if mpv_parse_response(mpv, object) {
+		return
+	}
+
+	event_value, has_event := object["event"]
+	if !has_event {
+		return
+	}
+	event_name, event_ok := event_value.(json.String)
+	if !event_ok {
+		return
+	}
+	if mpv.client != nil {
+		client_on_mpv_event(mpv.client, string(event_name), object)
+	}
+}
+
+mpv_reader_loop :: proc(data: rawptr) {
+	defer runtime.default_temp_allocator_destroy(auto_cast context.temp_allocator.data)
+	mpv := cast(^Mpv_Connection)data
+	framer: Line_Framer
+	framer_init(&framer)
+	defer framer_destroy(&framer)
+	buffer: [2048]byte
+
+	for {
+		count := posix.recv(mpv.fd, raw_data(buffer[:]), len(buffer), {})
+		if count <= 0 {
+			break
+		}
+		frames, frame_err := framer_push(&framer, buffer[:int(count)])
+		if frame_err != .None {
+			destroy_frames(frames)
+			break
+		}
+		for frame in frames {
+			if len(frame) > 0 {
+				mpv_handle_line(mpv, frame)
+			}
+		}
+		destroy_frames(frames)
+	}
+
+	sync.mutex_lock(&mpv.response_mutex)
+	mpv.closed = true
+	sync.cond_broadcast(&mpv.response_cond)
+	sync.mutex_unlock(&mpv.response_mutex)
+	if mpv.client != nil {
+		client_on_mpv_disconnect(mpv.client)
+	}
+}
+
+mpv_start_reader :: proc(mpv: ^Mpv_Connection) -> ^thread.Thread {
+	return thread.create_and_start_with_data(
+		rawptr(mpv),
+		mpv_reader_loop,
+		init_context = context,
+		self_cleanup = false,
+	)
+}
+
+mpv_close :: proc(mpv: ^Mpv_Connection) {
+	if mpv.fd >= 0 {
+		_ = posix.shutdown(mpv.fd, .RDWR)
+		_ = posix.close(mpv.fd)
+		mpv.fd = -1
+	}
+}

mpv_test.odin

@@ -0,0 +1,100 @@
+package main
+
+import "base:runtime"
+import "core:fmt"
+import "core:strings"
+import "core:sys/posix"
+import "core:testing"
+import "core:thread"
+
+Fake_Mpv :: struct {
+	fd:       posix.FD,
+	received: [3]string,
+	fail_seek: bool,
+}
+
+fake_mpv_read_line :: proc(fd: posix.FD, buffer: []byte) -> (line: string, ok: bool) {
+	length := 0
+	for length < len(buffer) {
+		count := posix.recv(fd, raw_data(buffer[length:]), 1, {})
+		if count != 1 {
+			return "", false
+		}
+		if buffer[length] == '\n' {
+			return string(buffer[:length]), true
+		}
+		length += 1
+	}
+	return "", false
+}
+
+fake_mpv_loop :: proc(data: rawptr) {
+	defer runtime.default_temp_allocator_destroy(auto_cast context.temp_allocator.data)
+	fake := cast(^Fake_Mpv)data
+	buffer: [512]byte
+	for index in 0..<3 {
+		line, ok := fake_mpv_read_line(fake.fd, buffer[:])
+		if !ok {
+			return
+		}
+		fake.received[index] = strings.clone(line)
+		error_name := "failure" if fake.fail_seek && index == 1 else "success"
+		response := fmt.aprintf(
+			"{{\"error\":\"%s\",\"request_id\":%d}}\n",
+			error_name,
+			index+1,
+		)
+		_ = mpv_send_all(fake.fd, response)
+		delete(response)
+	}
+}
+
+run_fake_seek_transaction :: proc(t: ^testing.T, fail_seek: bool) -> (ok: bool, fake: Fake_Mpv) {
+	fds: [2]posix.FD
+	testing.expect_value(t, posix.socketpair(.UNIX, .STREAM, .IP, &fds), posix.result.OK)
+
+	mpv := Mpv_Connection{fd = fds[0]}
+	reader := mpv_start_reader(&mpv)
+	testing.expect(t, reader != nil)
+	fake = Fake_Mpv{fd = fds[1], fail_seek = fail_seek}
+	fake_thread := thread.create_and_start_with_data(
+		rawptr(&fake),
+		fake_mpv_loop,
+		init_context = context,
+		self_cleanup = false,
+	)
+	testing.expect(t, fake_thread != nil)
+
+	ok = mpv_apply_remote_seek(&mpv, 42.5)
+	thread.join(fake_thread)
+	thread.destroy(fake_thread)
+	mpv_close(&mpv)
+	thread.join(reader)
+	thread.destroy(reader)
+	_ = posix.close(fake.fd)
+	return
+}
+
+@(test)
+remote_seek_masks_and_restores_seek_events :: proc(t: ^testing.T) {
+	ok, fake := run_fake_seek_transaction(t, false)
+	testing.expect(t, ok)
+
+	testing.expect(t, strings.contains(fake.received[0], `"disable_event","seek"`))
+	testing.expect(t, strings.contains(fake.received[1], `"seek",42.500000,"absolute+exact"`))
+	testing.expect(t, strings.contains(fake.received[2], `"enable_event","seek"`))
+	for line in fake.received {
+		delete(line)
+	}
+}
+
+@(test)
+remote_seek_restores_events_after_seek_failure :: proc(t: ^testing.T) {
+	ok, fake := run_fake_seek_transaction(t, true)
+	testing.expect(t, !ok)
+	testing.expect(t, strings.contains(fake.received[0], `"disable_event","seek"`))
+	testing.expect(t, strings.contains(fake.received[2], `"enable_event","seek"`))
+	for line in fake.received {
+		delete(line)
+	}
+}

protocol.odin

@@ -0,0 +1,278 @@
+package main
+
+import "core:encoding/json"
+import "core:fmt"
+import "core:math"
+import "core:strings"
+
+PROTOCOL_VERSION :: 1
+MAX_FRAME_BYTES  :: 4096
+
+Event_Kind :: enum {
+	Pause,
+	Seek,
+}
+
+Playback_Event :: struct {
+	kind:     Event_Kind,
+	paused:   bool,
+	position: f64,
+	client_id: u64,
+}
+
+Protocol_Error :: enum {
+	None,
+	Invalid_JSON,
+	Expected_Object,
+	Missing_Version,
+	Unsupported_Version,
+	Missing_Type,
+	Unknown_Type,
+	Missing_Paused,
+	Invalid_Paused,
+	Missing_Position,
+	Invalid_Position,
+	Missing_Client_ID,
+	Invalid_Client_ID,
+}
+
+protocol_error_string :: proc(err: Protocol_Error) -> string {
+	switch err {
+	case .None:                return "no error"
+	case .Invalid_JSON:        return "invalid JSON"
+	case .Expected_Object:     return "expected a JSON object"
+	case .Missing_Version:     return "missing integer version"
+	case .Unsupported_Version: return "unsupported protocol version"
+	case .Missing_Type:        return "missing string type"
+	case .Unknown_Type:        return "unknown event type"
+	case .Missing_Paused:      return "missing paused field"
+	case .Invalid_Paused:      return "paused must be a boolean"
+	case .Missing_Position:    return "missing position field"
+	case .Invalid_Position:    return "position must be a finite non-negative number"
+	case .Missing_Client_ID:   return "missing client_id field"
+	case .Invalid_Client_ID:   return "client_id must be a positive integer"
+	}
+	return "unknown protocol error"
+}
+
+parse_protocol_event :: proc(data: []byte) -> (event: Playback_Event, err: Protocol_Error) {
+	root: json.Value
+	if json.unmarshal(data, &root, spec = .JSON) != nil {
+		return {}, .Invalid_JSON
+	}
+	defer json.destroy_value(root)
+
+	object, ok := root.(json.Object)
+	if !ok {
+		return {}, .Expected_Object
+	}
+
+	version_value, found := object["version"]
+	if !found {
+		return {}, .Missing_Version
+	}
+	version, version_ok := version_value.(json.Integer)
+	if !version_ok {
+		return {}, .Missing_Version
+	}
+	if version != PROTOCOL_VERSION {
+		return {}, .Unsupported_Version
+	}
+
+	type_value, type_found := object["type"]
+	if !type_found {
+		return {}, .Missing_Type
+	}
+	type_name, type_ok := type_value.(json.String)
+	if !type_ok {
+		return {}, .Missing_Type
+	}
+
+	client_id: u64
+	if client_id_value, client_id_found := object["client_id"]; client_id_found {
+		parsed_id, client_id_ok := client_id_value.(json.Integer)
+		if !client_id_ok || parsed_id <= 0 {
+			return {}, .Invalid_Client_ID
+		}
+		client_id = u64(parsed_id)
+	}
+
+	switch string(type_name) {
+	case "pause":
+		paused_value, paused_found := object["paused"]
+		if !paused_found {
+			return {}, .Missing_Paused
+		}
+		paused, paused_ok := paused_value.(json.Boolean)
+		if !paused_ok {
+			return {}, .Invalid_Paused
+		}
+		return Playback_Event{kind = .Pause, paused = bool(paused), client_id = client_id}, .None
+
+	case "seek":
+		position_value, position_found := object["position"]
+		if !position_found {
+			return {}, .Missing_Position
+		}
+		position: f64
+		#partial switch value in position_value {
+		case json.Integer:
+			position = f64(value)
+		case json.Float:
+			position = f64(value)
+		case:
+			return {}, .Invalid_Position
+		}
+		if position < 0 || math.is_nan(position) || math.is_inf(position) {
+			return {}, .Invalid_Position
+		}
+		return Playback_Event{kind = .Seek, position = position, client_id = client_id}, .None
+	}
+
+	return {}, .Unknown_Type
+}
+
+encode_protocol_event :: proc(event: Playback_Event, allocator := context.allocator) -> (line: string) {
+	switch event.kind {
+	case .Pause:
+		if event.client_id > 0 {
+			return fmt.aprintf(
+				"{{\"version\":%d,\"type\":\"pause\",\"paused\":%s,\"client_id\":%d}}\n",
+				PROTOCOL_VERSION,
+				"true" if event.paused else "false",
+				event.client_id,
+				allocator = allocator,
+			)
+		}
+		return fmt.aprintf(
+			"{{\"version\":%d,\"type\":\"pause\",\"paused\":%s}}\n",
+			PROTOCOL_VERSION,
+			"true" if event.paused else "false",
+			allocator = allocator,
+		)
+	case .Seek:
+		if event.client_id > 0 {
+			return fmt.aprintf(
+				"{{\"version\":%d,\"type\":\"seek\",\"position\":%.6f,\"client_id\":%d}}\n",
+				PROTOCOL_VERSION,
+				event.position,
+				event.client_id,
+				allocator = allocator,
+			)
+		}
+		return fmt.aprintf(
+			"{{\"version\":%d,\"type\":\"seek\",\"position\":%.6f}}\n",
+			PROTOCOL_VERSION,
+			event.position,
+			allocator = allocator,
+		)
+	}
+	return strings.clone("", allocator)
+}
+
+encode_welcome :: proc(client_id: u64, allocator := context.allocator) -> string {
+	return fmt.aprintf(
+		"{{\"version\":%d,\"type\":\"welcome\",\"client_id\":%d}}\n",
+		PROTOCOL_VERSION,
+		client_id,
+		allocator = allocator,
+	)
+}
+
+parse_welcome :: proc(data: []byte) -> (client_id: u64, matched: bool, err: Protocol_Error) {
+	root: json.Value
+	if json.unmarshal(data, &root, spec = .JSON) != nil {
+		return 0, false, .Invalid_JSON
+	}
+	defer json.destroy_value(root)
+	object, object_ok := root.(json.Object)
+	if !object_ok {
+		return 0, false, .Expected_Object
+	}
+	type_value, type_found := object["type"]
+	if !type_found {
+		return 0, false, .Missing_Type
+	}
+	type_name, type_ok := type_value.(json.String)
+	if !type_ok {
+		return 0, false, .Missing_Type
+	}
+	if string(type_name) != "welcome" {
+		return 0, false, .None
+	}
+	matched = true
+	version_value, version_found := object["version"]
+	if !version_found {
+		return 0, true, .Missing_Version
+	}
+	version, version_ok := version_value.(json.Integer)
+	if !version_ok {
+		return 0, true, .Missing_Version
+	}
+	if version != PROTOCOL_VERSION {
+		return 0, true, .Unsupported_Version
+	}
+	client_id_value, client_id_found := object["client_id"]
+	if !client_id_found {
+		return 0, true, .Missing_Client_ID
+	}
+	parsed_id, client_id_ok := client_id_value.(json.Integer)
+	if !client_id_ok || parsed_id <= 0 {
+		return 0, true, .Invalid_Client_ID
+	}
+	return u64(parsed_id), true, .None
+}
+
+format_playback_log :: proc(
+	scope: string,
+	client_id: u64,
+	event: Playback_Event,
+	is_local := false,
+	allocator := context.allocator,
+) -> string {
+	if is_local {
+		switch event.kind {
+		case .Pause:
+			return fmt.aprintf(
+				"%s: client %d (you) %s",
+				scope,
+				client_id,
+				"paused" if event.paused else "played",
+				allocator = allocator,
+			)
+		case .Seek:
+			return fmt.aprintf(
+				"%s: client %d (you) sought to %.3f seconds",
+				scope,
+				client_id,
+				event.position,
+				allocator = allocator,
+			)
+		}
+	}
+	switch event.kind {
+	case .Pause:
+		return fmt.aprintf(
+			"%s: client %d %s",
+			scope,
+			client_id,
+			"paused" if event.paused else "played",
+			allocator = allocator,
+		)
+	case .Seek:
+		return fmt.aprintf(
+			"%s: client %d sought to %.3f seconds",
+			scope,
+			client_id,
+			event.position,
+			allocator = allocator,
+		)
+	}
+	return strings.clone("", allocator)
+}
+
+log_playback_event :: proc(scope: string, client_id: u64, event: Playback_Event, is_local := false) {
+	line := format_playback_log(scope, client_id, event, is_local)
+	defer delete(line)
+	fmt.println(line)
+}

protocol_test.odin

@@ -0,0 +1,73 @@
+package main
+
+import "core:strings"
+import "core:testing"
+
+@(test)
+protocol_round_trip_pause :: proc(t: ^testing.T) {
+	line := encode_protocol_event(Playback_Event{kind = .Pause, paused = true})
+	defer delete(line)
+	event, err := parse_protocol_event(transmute([]byte)strings.trim_space(line))
+	testing.expect_value(t, err, Protocol_Error.None)
+	testing.expect_value(t, event.kind, Event_Kind.Pause)
+	testing.expect_value(t, event.paused, true)
+}
+
+@(test)
+protocol_round_trip_seek :: proc(t: ^testing.T) {
+	line := encode_protocol_event(Playback_Event{kind = .Seek, position = 123.456})
+	defer delete(line)
+	event, err := parse_protocol_event(transmute([]byte)strings.trim_space(line))
+	testing.expect_value(t, err, Protocol_Error.None)
+	testing.expect_value(t, event.kind, Event_Kind.Seek)
+	testing.expect(t, event.position > 123.455 && event.position < 123.457)
+}
+
+@(test)
+protocol_rejects_invalid_messages :: proc(t: ^testing.T) {
+	missing_pause: string = `{"version":1,"type":"pause"}`
+	bad_version: string = `{"version":2,"type":"pause","paused":true}`
+	bad_position: string = `{"version":1,"type":"seek","position":-1}`
+	_, err := parse_protocol_event(transmute([]byte)missing_pause)
+	testing.expect_value(t, err, Protocol_Error.Missing_Paused)
+	_, err = parse_protocol_event(transmute([]byte)bad_version)
+	testing.expect_value(t, err, Protocol_Error.Unsupported_Version)
+	_, err = parse_protocol_event(transmute([]byte)bad_position)
+	testing.expect_value(t, err, Protocol_Error.Invalid_Position)
+}
+
+@(test)
+protocol_round_trip_sourced_event :: proc(t: ^testing.T) {
+	line := encode_protocol_event(Playback_Event{
+		kind = .Seek,
+		position = 9.25,
+		client_id = 7,
+	})
+	defer delete(line)
+	event, err := parse_protocol_event(transmute([]byte)strings.trim_space(line))
+	testing.expect_value(t, err, Protocol_Error.None)
+	testing.expect_value(t, event.client_id, u64(7))
+}
+
+@(test)
+protocol_parses_welcome :: proc(t: ^testing.T) {
+	line := encode_welcome(12)
+	defer delete(line)
+	client_id, matched, err := parse_welcome(transmute([]byte)strings.trim_space(line))
+	testing.expect(t, matched)
+	testing.expect_value(t, err, Protocol_Error.None)
+	testing.expect_value(t, client_id, u64(12))
+}
+
+@(test)
+playback_logs_identify_client_and_action :: proc(t: ^testing.T) {
+	paused := format_playback_log("server", 2, Playback_Event{kind = .Pause, paused = true})
+	defer delete(paused)
+	played := format_playback_log("client", 3, Playback_Event{kind = .Pause}, is_local = true)
+	defer delete(played)
+	sought := format_playback_log("client", 4, Playback_Event{kind = .Seek, position = 65.125})
+	defer delete(sought)
+	testing.expect_value(t, paused, "server: client 2 paused")
+	testing.expect_value(t, played, "client: client 3 (you) played")
+	testing.expect_value(t, sought, "client: client 4 sought to 65.125 seconds")
+}

server.odin

@@ -0,0 +1,196 @@
+package main
+
+import "base:runtime"
+import "core:fmt"
+import "core:net"
+import "core:sync"
+import "core:thread"
+import "core:sys/posix"
+
+MAX_SERVER_CLIENTS :: 16
+
+Server_State :: struct {
+	mutex:   sync.Mutex,
+	clients: [MAX_SERVER_CLIENTS]Server_Client,
+	next_id: u64,
+}
+
+Server_Client :: struct {
+	state:    ^Server_State,
+	socket:   net.TCP_Socket,
+	endpoint: net.Endpoint,
+	id:       u64,
+	active:   bool,
+}
+
+server_disconnect_locked :: proc(client: ^Server_Client, reason: string) {
+	if !client.active {
+		return
+	}
+	client.active = false
+	_ = net.shutdown(client.socket, .Both)
+	net.close(client.socket)
+	fmt.printfln("server: client %d disconnected (%s)", client.id, reason)
+}
+
+server_broadcast :: proc(sender: ^Server_Client, event: Playback_Event) {
+	state := sender.state
+	sync.mutex_lock(&state.mutex)
+	defer sync.mutex_unlock(&state.mutex)
+	if !sender.active {
+		return
+	}
+	outbound_event := event
+	outbound_event.client_id = sender.id
+	log_playback_event("server", sender.id, outbound_event)
+	line := encode_protocol_event(outbound_event)
+	defer delete(line)
+
+	for &client in state.clients {
+		if !client.active || client.id == sender.id {
+			continue
+		}
+		written, send_err := net.send_tcp(client.socket, transmute([]byte)line)
+		if send_err != nil || written != len(line) {
+			server_disconnect_locked(&client, "send failed")
+		}
+	}
+}
+
+server_client_loop :: proc(data: rawptr) {
+	defer runtime.default_temp_allocator_destroy(auto_cast context.temp_allocator.data)
+	client := cast(^Server_Client)data
+	framer: Line_Framer
+	framer_init(&framer)
+	defer framer_destroy(&framer)
+	buf: [2048]byte
+	disconnect_reason := "connection closed"
+
+	for {
+		count, recv_err := net.recv_tcp(client.socket, buf[:])
+		if recv_err != nil {
+			disconnect_reason = "receive failed"
+			break
+		}
+		if count == 0 {
+			break
+		}
+
+		frames, frame_err := framer_push(&framer, buf[:count])
+		if frame_err != .None {
+			destroy_frames(frames)
+			disconnect_reason = "oversized message"
+			break
+		}
+
+		valid := true
+		for frame in frames {
+			if len(frame) == 0 {
+				continue
+			}
+			event, protocol_err := parse_protocol_event(transmute([]byte)frame)
+			if protocol_err != .None {
+				fmt.printfln(
+					"server: client %d sent invalid message: %s",
+					client.id,
+					protocol_error_string(protocol_err),
+				)
+				disconnect_reason = "invalid message"
+				valid = false
+				break
+			}
+			server_broadcast(client, event)
+		}
+		destroy_frames(frames)
+		if !valid {
+			break
+		}
+	}
+
+	sync.mutex_lock(&client.state.mutex)
+	server_disconnect_locked(client, disconnect_reason)
+	sync.mutex_unlock(&client.state.mutex)
+}
+
+run_server :: proc(options: Options) -> bool {
+	address, ok := net.parse_ip4_address(options.host)
+	if !ok {
+		fmt.eprintfln("server: host must be an IPv4 address: %s", options.host)
+		return false
+	}
+	listener, listen_err := net.listen_tcp(net.Endpoint{address = address, port = options.port})
+	if listen_err != nil {
+		fmt.eprintfln("server: could not listen on %s:%d: %v", options.host, options.port, listen_err)
+		return false
+	}
+	register_server_listener(posix.FD(net.Socket(listener)))
+	defer clear_server_listener()
+
+	state := new(Server_State)
+	fmt.printfln("server: listening on %s:%d", options.host, options.port)
+
+	for {
+		socket, endpoint, accept_err := net.accept_tcp(listener)
+		if accept_err != nil {
+			if should_shutdown() {
+				fmt.println("server: shutdown requested")
+				return true
+			}
+			fmt.eprintfln("server: accept failed: %v", accept_err)
+			net.close(listener)
+			return false
+		}
+
+		client: ^Server_Client
+		client_ready := false
+		sync.mutex_lock(&state.mutex)
+		for &candidate in state.clients {
+			if !candidate.active {
+				client = &candidate
+				break
+			}
+		}
+		if client != nil {
+			state.next_id += 1
+			client^ = Server_Client{
+				state = state,
+				socket = socket,
+				endpoint = endpoint,
+				id = state.next_id,
+				active = true,
+			}
+			welcome := encode_welcome(client.id)
+			welcome_length := len(welcome)
+			written, welcome_err := net.send_tcp(client.socket, transmute([]byte)welcome)
+			delete(welcome)
+			if welcome_err == nil && written == welcome_length {
+				client_ready = true
+			} else {
+				server_disconnect_locked(client, "could not send welcome")
+			}
+		}
+		sync.mutex_unlock(&state.mutex)
+
+		if client == nil {
+			fmt.eprintln("server: connection rejected: client limit reached")
+			net.close(socket)
+			continue
+		}
+		if !client_ready {
+			continue
+		}
+
+		remote := net.endpoint_to_string(endpoint)
+		fmt.printfln("server: client %d connected from %s", client.id, remote)
+		if thread.create_and_start_with_data(
+			rawptr(client),
+			server_client_loop,
+			init_context = context,
+			self_cleanup = true,
+		) == nil {
+			sync.mutex_lock(&state.mutex)
+			server_disconnect_locked(client, "could not start client thread")
+			sync.mutex_unlock(&state.mutex)
+		}
+	}
+}

shutdown.odin

@@ -0,0 +1,34 @@
+package main
+
+import "base:intrinsics"
+import "core:c"
+import "core:c/libc"
+import "core:sys/posix"
+
+shutdown_requested: libc.sig_atomic_t
+server_listener_fd: libc.sig_atomic_t = -1
+
+shutdown_signal_handler :: proc "c" (signal: c.int) {
+	intrinsics.atomic_store(&shutdown_requested, 1)
+	fd := intrinsics.atomic_exchange(&server_listener_fd, libc.sig_atomic_t(-1))
+	if fd >= 0 {
+		_ = posix.close(posix.FD(fd))
+	}
+}
+
+register_server_listener :: proc(fd: posix.FD) {
+	intrinsics.atomic_store(&server_listener_fd, libc.sig_atomic_t(fd))
+}
+
+clear_server_listener :: proc() {
+	intrinsics.atomic_store(&server_listener_fd, libc.sig_atomic_t(-1))
+}
+
+install_shutdown_handlers :: proc() {
+	libc.signal(libc.SIGINT, shutdown_signal_handler)
+	libc.signal(libc.SIGTERM, shutdown_signal_handler)
+}
+
+should_shutdown :: proc() -> bool {
+	return intrinsics.atomic_load(&shutdown_requested) == 1
+}