refactor: split transfer and protocol responsibilities

src/server/server.c

@@ -1,24 +1,20 @@
-#include "array_list.h"
-#include "chunk.h"
 #include "config.h"
-#include "data.h"
 #include "file.h"
 #include "log.h"
 #include "multiprocessing.h"
-#include "protocol.h"
 #include "queue.h"
+#include "receiver.h"
 #include "transport_tcp.h"
 #include "transport_tls.h"
 #include "unistd.h"
 #include "utils.h"
+#include <fcntl.h>
+#include <limits.h>
 #include <signal.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
-#include <limits.h>
-#include <fcntl.h>
 #include <sys/stat.h>
-#include <unistd.h>
 
 static char* authorized_root;
 static int authorized_root_fd = -1;
@@ -29,11 +25,6 @@ static bool path_is_within(const char* root, const char* path) {
   return strncmp(root, path, n) == 0 && (path[n] == '\0' || path[n] == '/');
 }
 
-static bool valid_batch_path(const char* path) {
-  return path && path[0] != '\0' && path[0] != '/' && !has_path_traversal(path) &&
-         strchr(path, '\0') == path + strlen(path);
-}
-
 static bool __attribute__((unused)) configure_authorization(const char* root) {
   char resolved[PATH_MAX];
   if (!root || !realpath(root, resolved))
@@ -52,131 +43,24 @@ static bool __attribute__((unused)) configure_authorization(const char* root) {
   return true;
 }
 
-int receive_files(Config* config, int fd) {
-  Status status;
-  if (!receive_status(fd, &status))
-    return -1;
-
-  while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK ||
-         status == STATUS_KEEPALIVE || status == STATUS_ABORT || status == STATUS_CHECK_BATCH) {
-    if (status == STATUS_KEEPALIVE) {
-      send_status(fd, STATUS_KEEPALIVE);
-      goto next;
-    }
-    if (status == STATUS_ABORT) {
-      log_message(LOG_LEVEL_INFO, "Received abort from client, cleaning up");
-      return -1;
-    }
-    if (status == STATUS_CHECK) {
-      bool skipped;
-      File* file = receive_incremental_check(fd, config, &skipped);
-      if (skipped)
-        goto next;
-      if (file == NULL && !skipped)
-        return -1;
-      if (config->save_to_disk &&
-          !file_save_to_disk(config->receive_root_directory, file, config)) {
-        file_destroy(file);
-        send_status(fd, STATUS_ERROR);
-        return -1;
-      }
-      file_destroy(file);
-    } else if (status == STATUS_CHUNK) {
-      Chunk* chunk = receive_chunk_data(fd, config);
-      if (chunk == NULL) {
-        send_status(fd, STATUS_ERROR);
-        return -1;
-      }
-      for (int i = 0; i < chunk->element_count; i++) {
-        if (config->save_to_disk &&
-            !file_save_to_disk(config->receive_root_directory, chunk->items[i], config)) {
-          chunk_destroy(chunk);
-          send_status(fd, STATUS_ERROR);
-          return -1;
-        }
-      }
-      chunk_destroy(chunk);
-    } else if (status == STATUS_CHECK_BATCH) {
-      int count;
-      /* Batch framing has no checksum field yet; never silently downgrade a
-         checksum-enabled transfer into mtime-only matching. */
-      if (config->checksum || !receive_int(fd, &count) || count < 0 || count > MAX_MANIFEST_ENTRIES)
-        return -1;
-      for (int i = 0; i < count; i++) {
-        char* check_path = receive_str(fd);
-        if (!check_path)
-          return -1;
-        unsigned long long check_size;
-        long long check_mtime;
-        if (!receive_n_data(fd, &check_size, sizeof(check_size)) ||
-            !receive_n_data(fd, &check_mtime, sizeof(check_mtime))) {
-          free(check_path);
-          return -1;
-        }
-        if (!valid_batch_path(check_path)) {
-          free(check_path);
-          send_status(fd, STATUS_ERROR);
-          return -1;
-        }
-        char* full_path = path_cat(config->receive_root_directory, check_path);
-        struct stat st;
-        bool has_old = full_path && lstat(full_path, &st) == 0;
-        bool match = has_old && (unsigned long long)st.st_size == check_size &&
-                     (long long)st.st_mtime == check_mtime;
-        bool sent = send_status(fd, match ? STATUS_OK : STATUS_NEXT);
-        free(full_path);
-        free(check_path);
-        if (!sent)
-          return -1;
-      }
-      goto next;
-    } else {
-      File* file = file_receive(config, fd);
-      if (file == NULL) {
-        log_message(LOG_LEVEL_ERROR, "Failed to receive file");
-        send_status(fd, STATUS_ERROR);
-        return -1;
-      }
-      if (config->save_to_disk &&
-          !file_save_to_disk(config->receive_root_directory, file, config)) {
-        file_destroy(file);
-        send_status(fd, STATUS_ERROR);
-        return -1;
-      }
-      file_destroy(file);
-    }
-  next:
-    if (!receive_status(fd, &status)) {
-      send_status(fd, STATUS_ERROR);
-      return -1;
-    }
-  }
-
-  if (status == STATUS_MANIFEST) {
-    if (receive_manifest(fd, config, &status) != 0)
-      return -1;
-  }
-  if (status != STATUS_FINISHED) {
-    log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED Status");
-    send_status(fd, STATUS_ERROR);
-    return -1;
-  }
-  send_status(fd, STATUS_OK);
-  return 0;
-}
-
 void handler(int file_descriptor) {
   SSL* ssl = io_get_ssl();
+  ProtocolSession session;
+  protocol_session_init(&session, file_descriptor, file_descriptor);
+  protocol_session_set_ssl(&session, ssl);
+  protocol_session_bind(&session);
   Config* config = config_receive(file_descriptor);
   if (config == NULL) {
     log_message(LOG_LEVEL_ERROR, "Failed to receive config");
     close(file_descriptor);
+    protocol_session_unbind();
     return;
   }
   if (!authorized_root) {
     log_message(LOG_LEVEL_ERROR, "No server-side destination root configured");
     config_delete(config);
     close(file_descriptor);
+    protocol_session_unbind();
     return;
   }
   char resolved_destination[PATH_MAX];
@@ -188,6 +72,7 @@ void handler(int file_descriptor) {
     free(canonical_destination);
     config_delete(config);
     close(file_descriptor);
+    protocol_session_unbind();
     return;
   }
   if (canonical_destination)
@@ -200,6 +85,7 @@ void handler(int file_descriptor) {
   if (!config->receive_root_directory) {
     config_delete(config);
     close(file_descriptor);
+    protocol_session_unbind();
     return;
   }
   config->use_delete = config->use_delete && allow_delete;
@@ -208,6 +94,7 @@ void handler(int file_descriptor) {
     if (q == NULL) {
       config_delete(config);
       close(file_descriptor);
+      protocol_session_unbind();
       return;
     }
     PipelineContextReceiver* context =
@@ -216,14 +103,14 @@ void handler(int file_descriptor) {
       queue_destroy(q);
       config_delete(config);
       close(file_descriptor);
+      protocol_session_unbind();
       return;
     }
     thrd_t receiver, writer;
-    bool receiver_created = false;
+    bool receiver_created = thrd_create(&receiver, receive_thread, context) == thrd_success;
     bool writer_created = false;
-    receiver_created = (thrd_create(&receiver, receive_thread, context) == thrd_success);
     if (receiver_created)
-      writer_created = (thrd_create(&writer, write_thread, context) == thrd_success);
+      writer_created = thrd_create(&writer, write_thread, context) == thrd_success;
     if (!receiver_created || !writer_created) {
       perror("Error creating Threads");
       if (receiver_created) {
@@ -240,22 +127,23 @@ void handler(int file_descriptor) {
       if (writer_created)
         thrd_join(writer, NULL);
       pipeline_context_receiver_destroy(context);
+      protocol_session_unbind();
       return;
     }
     int receiver_result;
     int writer_result;
     thrd_join(receiver, &receiver_result);
     thrd_join(writer, &writer_result);
-    if (receiver_result == thrd_success && writer_result == thrd_success)
-      send_status(file_descriptor, STATUS_OK);
-    else
-      send_status(file_descriptor, STATUS_ERROR);
+    send_status(file_descriptor, receiver_result == thrd_success && writer_result == thrd_success
+                                     ? STATUS_OK
+                                     : STATUS_ERROR);
     pipeline_context_receiver_destroy(context);
   } else {
-    if (receive_files(config, file_descriptor) != 0)
+    if (receiver_receive_files(config, file_descriptor) != 0)
       log_message(LOG_LEVEL_ERROR, "Transfer failed");
     config_delete(config);
   }
+  protocol_session_unbind();
   close(file_descriptor);
 }
 
@@ -264,16 +152,14 @@ static Server* g_server = NULL;
 
 static void cleanup(int sig) {
   (void)sig;
-  if (g_server) {
+  if (g_server)
     server_delete(&g_server);
-  }
   _exit(0);
 }
 
 static void print_server_usage(void) {
   printf("FastSync Server\n");
-  printf("Usage: fastsync-server [options]\n");
-  printf("\n");
+  printf("Usage: fastsync-server [options]\n\n");
   printf("Options:\n");
   printf("  --stdio             Run in stdio mode (SSH transport)\n");
   printf("  -p <port>           TCP port (default: 8080, range: 1-65535)\n");
@@ -289,9 +175,7 @@ static void print_server_usage(void) {
 
 int main(int argc, char* argv[]) {
   bool use_tls = false;
-  char* tls_cert = NULL;
-  char* tls_key = NULL;
-  char* tls_ca = NULL;
+  char *tls_cert = NULL, *tls_key = NULL, *tls_ca = NULL;
   int port = 8080;
   const char* destination_root = ".";
   bool stdio_mode = false;
@@ -331,11 +215,8 @@ int main(int argc, char* argv[]) {
       return 1;
     }
   }
-
-  if (tls_ca && !use_tls) {
+  if (tls_ca && !use_tls)
     log_message(LOG_LEVEL_WARNING, "--ca has no effect without --tls");
-  }
-
   signal(SIGINT, cleanup);
   signal(SIGTERM, cleanup);
   if (!configure_authorization(destination_root)) {
@@ -352,7 +233,7 @@ int main(int argc, char* argv[]) {
     return 0;
   }
   g_server = server_create(port);
-  if (g_server == NULL) {
+  if (!g_server) {
     log_message(LOG_LEVEL_ERROR, "Failed to create server");
     return 1;
   }
@@ -374,4 +255,4 @@ int main(int argc, char* argv[]) {
   }
   return 0;
 }
-#endif /* !FASTSYNC_SERVER_AS_LIB */
+#endif