<commit_before>#include <assert.h>
#include <errno.h>
#include <math.h>
#include <stdio.h>
#include <string.h>
#include <strings.h>
#include <time.h>

#include <uuid/uuid.h>

#include "cosmopolite.h"

#define COSMO_CHECK_SECONDS 10

#define min(a, b) ((a) < (b) ? (a) : (b))
#define max(a, b) ((a) > (b) ? (a) : (b))

#define DELAY_MIN_MS 250
#define DELAY_MAX_MS 32000
#define DELAY_EXPONENT 1.1
#define DELAY_STAGGER_FACTOR 10

typedef struct {
  char *send_buf;
  size_t send_buf_len;

  char *recv_buf;
  size_t recv_buf_len;

  int64_t retry_after;
} cosmo_transfer;

static json_t *cosmo_find_subscription(cosmo *instance, json_t *subject) {
  size_t i;
  json_t *subscription;
  json_array_foreach(instance->subscriptions, i, subscription) {
    if (json_equal(json_object_get(subscription, "subject"), subject)) {
      return subscription;
    }
  }
  return NULL;
}

static size_t cosmo_read_callback(void *ptr, size_t size, size_t nmemb, void *userp) {
  cosmo_transfer *transfer = userp;
  size_t to_write = min(transfer->send_buf_len, size * nmemb);
  memcpy(ptr, transfer->send_buf, to_write);
  transfer->send_buf += to_write;
  transfer->send_buf_len -= to_write;
  return to_write;
}

static size_t cosmo_write_callback(void *ptr, size_t size, size_t nmemb, void *userp) {
  cosmo_transfer *transfer = userp;
  size_t to_read = size * nmemb;
  transfer->recv_buf = realloc(transfer->recv_buf, transfer->recv_buf_len + to_read + 1);
  assert(transfer->recv_buf);
  memcpy(transfer->recv_buf + transfer->recv_buf_len, ptr, to_read);
  transfer->recv_buf_len += to_read;
  transfer->recv_buf[transfer->recv_buf_len] = '\0';
  return to_read;
}

static size_t cosmo_header_callback(char *ptr, size_t size, size_t nmemb, void *userp) {
  cosmo_transfer *transfer = userp;
  size_t length = size * nmemb;
#define RETRY_AFTER_HEADER "Retry-After: 0\r\n"
#define RETRY_AFTER_HEADER_SIZE (sizeof(RETRY_AFTER_HEADER) - 1)
  if (length == RETRY_AFTER_HEADER_SIZE &&
      strncasecmp(ptr, RETRY_AFTER_HEADER, RETRY_AFTER_HEADER_SIZE) == 0) {
    transfer->retry_after = 0;
  }
  return length;
}

static char *cosmo_build_rpc(const cosmo *instance, const json_t *commands) {
  json_t *to_send = json_pack("{sssssO}", "client_id", instance->client_id, "instance_id", instance->instance_id, "commands", commands);
  assert(to_send);
  char *ret = json_dumps(to_send, 0);
  assert(ret);
  json_decref(to_send);
  return ret;
}

static bool cosmo_send_http_int(cosmo *instance, cosmo_transfer *transfer) {
  CURLcode res;
 
  assert(!curl_easy_setopt(instance->curl, CURLOPT_POSTFIELDSIZE, transfer->send_buf_len));
  assert(!curl_easy_setopt(instance->curl, CURLOPT_READDATA, transfer));
  assert(!curl_easy_setopt(instance->curl, CURLOPT_WRITEDATA, transfer));
  assert(!curl_easy_setopt(instance->curl, CURLOPT_HEADERDATA, transfer));
  res = curl_easy_perform(instance->curl);

  if (res) {
    fprintf(stderr, "curl_easy_perform() failed: %s\n", curl_easy_strerror(res));
    return false;
  }

  long return_code;
  assert(curl_easy_getinfo(instance->curl, CURLINFO_RESPONSE_CODE, &return_code) == CURLE_OK);
  if (return_code != 200) {
    fprintf(stderr, "server returned error: %ld\n", return_code);
    return false;
  }

  return true;
}

// Takes ownership of request.
static char *cosmo_send_http(cosmo *instance, char *request) {
  cosmo_transfer transfer = {
    .send_buf = request,
    .send_buf_len = strlen(request),
    .recv_buf = NULL,
    .recv_buf_len = 0,
    .retry_after = -1
  };

  int ret = cosmo_send_http_int(instance, &transfer);

  if (transfer.retry_after >= 0) {
    instance->next_delay_ms = transfer.retry_after * 1000;
  }

  free(request);

  return ret ? transfer.recv_buf : NULL;
}

static void cosmo_handle_message(cosmo *instance, json_t *event) {
  json_t *subject = json_object_get(event, "subject");
  if (!subject) {
    fprintf(stderr, "message event without subject\n");
    return;
  }

  json_t *id_json = json_object_get(event, "id");
  if (!id_json) {
    fprintf(stderr, "message event without id\n");
    return;
  }
  json_int_t id = json_integer_value(id_json);

  assert(!pthread_mutex_lock(&instance->lock));
  json_t *subscription = cosmo_find_subscription(instance, subject);
  if (!subscription) {
    fprintf(stderr, "message from unknown subject\n");
    assert(!pthread_mutex_unlock(&instance->lock));
    return;
  }
  json_t *messages = json_object_get(subscription, "messages");
  ssize_t insert_after;
  for (insert_after = json_array_size(messages) - 1; insert_after >= 0; insert_after--) {
    json_t *message = json_array_get(messages, insert_after);
    json_int_t message_id = json_integer_value(json_object_get(message, "id"));
    if (message_id == id) {
      assert(!pthread_mutex_unlock(&instance->lock));
      return;
    }
    if (message_id < id) {
      break;
    }
  }
  json_array_insert(messages, insert_after + 1, event);
  assert(!pthread_mutex_unlock(&instance->lock));

  if (instance->callbacks.message) {
    instance->callbacks.message(event);
  }
}

static void cosmo_handle_event(cosmo *instance, json_t *event) {
  const char *event_type = json_string_value(json_object_get(event, "event_type"));
  if (strcmp(event_type, "message") == 0) {
    cosmo_handle_message(instance, event);
  } else {
    fprintf(stderr, "unknown event type: %s\n", event_type);
  }<commit_msg>Decode incoming message JSON.<commit_after>#include <assert.h>
#include <errno.h>
#include <math.h>
#include <stdio.h>
#include <string.h>
#include <strings.h>
#include <time.h>

#include <uuid/uuid.h>

#include "cosmopolite.h"

#define COSMO_CHECK_SECONDS 10

#define min(a, b) ((a) < (b) ? (a) : (b))
#define max(a, b) ((a) > (b) ? (a) : (b))

#define DELAY_MIN_MS 250
#define DELAY_MAX_MS 32000
#define DELAY_EXPONENT 1.1
#define DELAY_STAGGER_FACTOR 10

typedef struct {
  char *send_buf;
  size_t send_buf_len;

  char *recv_buf;
  size_t recv_buf_len;

  int64_t retry_after;
} cosmo_transfer;

static json_t *cosmo_find_subscription(cosmo *instance, json_t *subject) {
  size_t i;
  json_t *subscription;
  json_array_foreach(instance->subscriptions, i, subscription) {
    if (json_equal(json_object_get(subscription, "subject"), subject)) {
      return subscription;
    }
  }
  return NULL;
}

static size_t cosmo_read_callback(void *ptr, size_t size, size_t nmemb, void *userp) {
  cosmo_transfer *transfer = userp;
  size_t to_write = min(transfer->send_buf_len, size * nmemb);
  memcpy(ptr, transfer->send_buf, to_write);
  transfer->send_buf += to_write;
  transfer->send_buf_len -= to_write;
  return to_write;
}

static size_t cosmo_write_callback(void *ptr, size_t size, size_t nmemb, void *userp) {
  cosmo_transfer *transfer = userp;
  size_t to_read = size * nmemb;
  transfer->recv_buf = realloc(transfer->recv_buf, transfer->recv_buf_len + to_read + 1);
  assert(transfer->recv_buf);
  memcpy(transfer->recv_buf + transfer->recv_buf_len, ptr, to_read);
  transfer->recv_buf_len += to_read;
  transfer->recv_buf[transfer->recv_buf_len] = '\0';
  return to_read;
}

static size_t cosmo_header_callback(char *ptr, size_t size, size_t nmemb, void *userp) {
  cosmo_transfer *transfer = userp;
  size_t length = size * nmemb;
#define RETRY_AFTER_HEADER "Retry-After: 0\r\n"
#define RETRY_AFTER_HEADER_SIZE (sizeof(RETRY_AFTER_HEADER) - 1)
  if (length == RETRY_AFTER_HEADER_SIZE &&
      strncasecmp(ptr, RETRY_AFTER_HEADER, RETRY_AFTER_HEADER_SIZE) == 0) {
    transfer->retry_after = 0;
  }
  return length;
}

static char *cosmo_build_rpc(const cosmo *instance, const json_t *commands) {
  json_t *to_send = json_pack("{sssssO}", "client_id", instance->client_id, "instance_id", instance->instance_id, "commands", commands);
  assert(to_send);
  char *ret = json_dumps(to_send, 0);
  assert(ret);
  json_decref(to_send);
  return ret;
}

static bool cosmo_send_http_int(cosmo *instance, cosmo_transfer *transfer) {
  CURLcode res;
 
  assert(!curl_easy_setopt(instance->curl, CURLOPT_POSTFIELDSIZE, transfer->send_buf_len));
  assert(!curl_easy_setopt(instance->curl, CURLOPT_READDATA, transfer));
  assert(!curl_easy_setopt(instance->curl, CURLOPT_WRITEDATA, transfer));
  assert(!curl_easy_setopt(instance->curl, CURLOPT_HEADERDATA, transfer));
  res = curl_easy_perform(instance->curl);

  if (res) {
    fprintf(stderr, "curl_easy_perform() failed: %s\n", curl_easy_strerror(res));
    return false;
  }

  long return_code;
  assert(curl_easy_getinfo(instance->curl, CURLINFO_RESPONSE_CODE, &return_code) == CURLE_OK);
  if (return_code != 200) {
    fprintf(stderr, "server returned error: %ld\n", return_code);
    return false;
  }

  return true;
}

// Takes ownership of request.
static char *cosmo_send_http(cosmo *instance, char *request) {
  cosmo_transfer transfer = {
    .send_buf = request,
    .send_buf_len = strlen(request),
    .recv_buf = NULL,
    .recv_buf_len = 0,
    .retry_after = -1
  };

  int ret = cosmo_send_http_int(instance, &transfer);

  if (transfer.retry_after >= 0) {
    instance->next_delay_ms = transfer.retry_after * 1000;
  }

  free(request);

  return ret ? transfer.recv_buf : NULL;
}

static void cosmo_handle_message(cosmo *instance, json_t *event) {
  json_t *subject = json_object_get(event, "subject");
  if (!subject) {
    fprintf(stderr, "message event without subject\n");
    return;
  }

  json_t *id_json = json_object_get(event, "id");
  if (!id_json) {
    fprintf(stderr, "message event without id\n");
    return;
  }
  json_int_t id = json_integer_value(id_json);

  json_t *message_content = json_object_get(event, "message");
  if (!message_content) {
    fprintf(stderr, "message event without content\n");
    return;
  }

  json_error_t err;
  json_t *message_object = json_loads(json_string_value(message_content), 0, &err);
  if (!message_object) {
    fprintf(stderr, "error parsing message content: %s\n", err.text);
    return;
  }
  json_object_set_new(event, "message", message_object);

  assert(!pthread_mutex_lock(&instance->lock));
  json_t *subscription = cosmo_find_subscription(instance, subject);
  if (!subscription) {
    fprintf(stderr, "message from unknown subject\n");
    assert(!pthread_mutex_unlock(&instance->lock));
    return;
  }

  json_t *messages = json_object_get(subscription, "messages");
  ssize_t insert_after;
  for (insert_after = json_array_size(messages) - 1; insert_after >= 0; insert_after--) {
    json_t *message = json_array_get(messages, insert_after);
    json_int_t message_id = json_integer_value(json_object_get(message, "id"));
    if (message_id == id) {
      assert(!pthread_mutex_unlock(&instance->lock));
      return;
    }
    if (message_id < id) {
      break;
    }
  }
  json_array_insert(messages, insert_after + 1, event);
  assert(!pthread_mutex_unlock(&instance->lock));

  if (instance->callbacks.message) {
    instance->callbacks.message(event);
  }
}

static void cosmo_handle_event(cosmo *instance, json_t *event) {
  const char *event_type = json_string_value(json_object_get(event, "event_type"));
  if (strcmp(event_type, "message") == 0) {
    cosmo_handle_message(instance, event);
  } else {
    fprintf(stderr, "unknown event type: %s\n", event_type);
  }
{'fraction_non_alphanumeric': 0.1304930332261522, 'fraction_numerical': 0.00669882100750268, 'mean_word_length': 4.877690288713911, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 87, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5864710', 'n_tokens_mistral': 4202, 'n_tokens_neox': 4068, 'n_words': 833}