MrDevBot/pretranscode
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
v2.0.0
pretranscode/pretranscode.c
T

1334 lines
56 KiB
C
Raw Permalink Blame History

#define _GNU_SOURCE
#include <curl/curl.h>
#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <fts.h>
#include <json-c/json.h>
#include <linux/ioprio.h>
#include <linux/sched.h>
#include <poll.h>
#include <sched.h>
#include <signal.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/file.h>
#include <sys/stat.h>
#include <sys/statvfs.h>
#include <sys/syscall.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>
#include <xxhash.h>
/* A single-writer, atomic Jellyfin compatibility cache. */
#define VERSION "2.0.0"
#define HASH_HEX 32
#define AUDIO_BITRATE "192k"
#define HASH_CHUNK (4U * 1024U * 1024U)
#define CAPTURE_MAX (16U * 1024U * 1024U)
#define ARRAY_LEN(a) (sizeof(a) / sizeof((a)[0]))
struct config {
char *source_root, *view_root, *cache_root, *state_root;
char *jellyfin_url, *library_root, *original_root, *jellyfin_cache_root;
char *api_key, *ffmpeg, *ffprobe;
char *crf, *preset, *maxrate, *bufsize, *profile;
char profile_hash[HASH_HEX + 1];
uint64_t cache_max;
unsigned cooldown_seconds, poll_seconds, rescan_seconds;
};
struct strings { char **v; size_t n, cap; };
struct buffer { char *v; size_t n, cap; };
struct timer { char *key; double value; };
struct timers { struct timer *v; size_t n, cap; };
struct media {
char codec[24], profile[64], transfer[32];
int depth, width, height, audio, subtitles, attachments, chapters;
double duration;
bool audio_compatible, dolby_vision;
};
struct catalog_item { char *relative, *id; struct media media; };
struct catalog { struct catalog_item *v; size_t n, cap; };
struct cache {
char *name, *path, *target;
char path_hash[HASH_HEX + 1];
char generation_hash[HASH_HEX + 1];
char profile_hash[HASH_HEX + 1];
char output_hash[HASH_HEX + 1];
char extension[5];
uint64_t size;
double used_at;
bool named, referenced, retain;
};
struct caches { struct cache *v; size_t n, cap; };
struct item {
char *relative, *source, *view;
char path_hash[HASH_HEX + 1];
char generation_hash[HASH_HEX + 1];
const char *extension;
uint64_t size;
double mtime;
struct catalog_item *catalog;
struct cache *cache, *fallback;
char *live_target;
};
struct items { struct item *v; size_t n, cap; };
struct manager {
struct config cfg;
struct strings active_items, active_keys, previous_items, previous_keys;
struct timers cooldowns, priorities, retries;
struct catalog catalog;
struct caches caches;
struct items items;
pid_t child;
int cache_lock;
bool sessions_healthy;
double last_scan;
};
static volatile sig_atomic_t stopping;
static void die(const char *fmt, ...) __attribute__((format(printf, 1, 2), noreturn));
static void logmsg(const char *fmt, ...) __attribute__((format(printf, 1, 2)));
static void die(const char *fmt, ...)
{
va_list ap;
va_start(ap, fmt);
fputs("pretranscode: ", stderr);
vfprintf(stderr, fmt, ap);
va_end(ap);
fputc('\n', stderr);
exit(EXIT_FAILURE);
}
static void logmsg(const char *fmt, ...)
{
char stamp[32];
time_t now = time(NULL);
struct tm tm;
gmtime_r(&now, &tm);
strftime(stamp, sizeof(stamp), "%Y-%m-%dT%H:%M:%SZ", &tm);
printf("%s ", stamp);
va_list ap;
va_start(ap, fmt);
vprintf(fmt, ap);
va_end(ap);
putchar('\n');
fflush(stdout);
}
static void *xmalloc(size_t n)
{
void *p = malloc(n ? n : 1);
if (!p) die("out of memory");
return p;
}
static void *xrealloc(void *p, size_t n)
{
p = realloc(p, n ? n : 1);
if (!p) die("out of memory");
return p;
}
static char *xstrdup(const char *s)
{
char *p = strdup(s ? s : "");
if (!p) die("out of memory");
return p;
}
static char *xprintf(const char *fmt, ...) __attribute__((format(printf, 1, 2)));
static char *xprintf(const char *fmt, ...)
{
char *s = NULL;
va_list ap;
va_start(ap, fmt);
if (vasprintf(&s, fmt, ap) < 0) s = NULL;
va_end(ap);
if (!s) die("out of memory");
return s;
}
static double seconds(clockid_t id)
{
struct timespec ts;
if (clock_gettime(id, &ts) < 0) die("clock_gettime: %s", strerror(errno));
return (double)ts.tv_sec + (double)ts.tv_nsec / 1e9;
}
static double now(void) { return seconds(CLOCK_REALTIME); }
static double monotonic(void) { return seconds(CLOCK_MONOTONIC); }
static bool ends(const char *s, const char *suffix)
{
size_t a = strlen(s), b = strlen(suffix);
return a >= b && memcmp(s + a - b, suffix, b) == 0;
}
static bool hex_hash(const char *s)
{
for (size_t i = 0; i < HASH_HEX; i++)
if (!((s[i] >= '0' && s[i] <= '9') || (s[i] >= 'a' && s[i] <= 'f')))
return false;
return s[HASH_HEX] == 0;
}
static bool relative_path(const char *s)
{
if (!s || !*s || *s == '/') return false;
for (const char *p = s; *p;) {
const char *e = strchr(p, '/');
size_t n = e ? (size_t)(e - p) : strlen(p);
if (!n || (n == 1 && p[0] == '.') || (n == 2 && p[0] == '.' && p[1] == '.'))
return false;
if (!e) break;
p = e + 1;
}
return true;
}
static bool video_name(const char *name)
{
static const char *const extensions[] = {
".mkv", ".mp4", ".m4v", ".avi", ".mov", ".ts", ".m2ts", ".webm"
};
const char *ext = strrchr(name, '.');
if (!ext) return false;
for (size_t i = 0; i < ARRAY_LEN(extensions); i++)
if (strcasecmp(ext, extensions[i]) == 0) return true;
return false;
}
static const char *cache_extension(const char *relative)
{
const char *ext = strrchr(relative, '.');
return ext && (strcasecmp(ext, ".mp4") == 0 || strcasecmp(ext, ".m4v") == 0)
? ".mp4" : ".mkv";
}
static int mkdirs(const char *path)
{
char *copy = xstrdup(path);
for (char *p = copy + 1; *p; p++) {
if (*p != '/') continue;
*p = 0;
if (mkdir(copy, 0755) < 0 && errno != EEXIST) {
int saved = errno; free(copy); errno = saved; return -1;
}
*p = '/';
}
int rc = mkdir(copy, 0755);
if (rc < 0 && errno == EEXIST) rc = 0;
int saved = errno; free(copy); errno = saved; return rc;
}
static int sync_directory(const char *path)
{
int fd = open(path, O_RDONLY | O_DIRECTORY | O_CLOEXEC);
if (fd < 0) return -1;
int rc = fsync(fd), saved = errno;
close(fd); errno = saved; return rc;
}
static int sync_parent(const char *path)
{
char *copy = xstrdup(path), *slash = strrchr(copy, '/');
const char *parent = ".";
if (slash) {
if (slash == copy) slash[1] = 0; else *slash = 0;
parent = copy;
}
int rc = sync_directory(parent), saved = errno;
free(copy); errno = saved; return rc;
}
static char *readlink_alloc(const char *path)
{
for (size_t cap = 256;; cap *= 2) {
char *s = xmalloc(cap + 1);
ssize_t n = readlink(path, s, cap);
if (n < 0) { free(s); return NULL; }
if ((size_t)n < cap) { s[n] = 0; return s; }
free(s);
}
}
static int atomic_link(const char *link, const char *target)
{
char *current = readlink_alloc(link);
if (current && strcmp(current, target) == 0) { free(current); return 0; }
free(current);
char *parent = xstrdup(link), *slash = strrchr(parent, '/');
if (!slash) { free(parent); errno = EINVAL; return -1; }
*slash = 0;
if (mkdirs(parent) < 0) { int e = errno; free(parent); errno = e; return -1; }
char *temporary = xprintf("%s/.pretranscode-link.%ld", parent, (long)getpid());
unlink(temporary);
int rc = -1;
if (symlink(target, temporary) == 0 && rename(temporary, link) == 0 && sync_parent(link) == 0)
rc = 1;
int saved = errno;
if (rc < 0) unlink(temporary);
free(temporary); free(parent); errno = saved; return rc;
}
static int string_cmp(const void *a, const void *b)
{
return strcmp(*(const char *const *)a, *(const char *const *)b);
}
static void strings_add(struct strings *s, const char *value)
{
if (s->n == s->cap) {
s->cap = s->cap ? s->cap * 2 : 16;
s->v = xrealloc(s->v, s->cap * sizeof(*s->v));
}
s->v[s->n++] = xstrdup(value);
}
static void strings_sort(struct strings *s)
{
qsort(s->v, s->n, sizeof(*s->v), string_cmp);
size_t out = 0;
for (size_t i = 0; i < s->n; i++) {
if (out && strcmp(s->v[i], s->v[out - 1]) == 0) free(s->v[i]);
else s->v[out++] = s->v[i];
}
s->n = out;
}
static bool strings_has(const struct strings *s, const char *value)
{
return bsearch(&value, s->v, s->n, sizeof(*s->v), string_cmp) != NULL;
}
static void strings_free(struct strings *s)
{
for (size_t i = 0; i < s->n; i++) free(s->v[i]);
free(s->v); memset(s, 0, sizeof(*s));
}
static void strings_move(struct strings *to, struct strings *from)
{
strings_free(to); *to = *from; memset(from, 0, sizeof(*from));
}
static void buffer_add(struct buffer *b, const void *data, size_t n)
{
if (n > CAPTURE_MAX - b->n) die("subprocess output exceeded limit");
if (b->n + n + 1 > b->cap) {
size_t cap = b->cap ? b->cap : 4096;
while (cap < b->n + n + 1) cap *= 2;
b->v = xrealloc(b->v, cap); b->cap = cap;
}
memcpy(b->v + b->n, data, n); b->n += n; b->v[b->n] = 0;
}
static size_t curl_write(void *data, size_t size, size_t count, void *opaque)
{
struct buffer *b = opaque;
size_t n = size * count;
if (n > CAPTURE_MAX - b->n) return 0;
buffer_add(b, data, n); return n;
}
static const char *jstring(json_object *o, const char *key, const char *fallback)
{
json_object *v = NULL;
return o && json_object_object_get_ex(o, key, &v) && json_object_is_type(v, json_type_string)
? json_object_get_string(v) : fallback;
}
static int jint(json_object *o, const char *key, int fallback)
{
json_object *v = NULL;
return o && json_object_object_get_ex(o, key, &v) ? json_object_get_int(v) : fallback;
}
static double jdouble(json_object *o, const char *key, double fallback)
{
json_object *v = NULL;
return o && json_object_object_get_ex(o, key, &v) ? json_object_get_double(v) : fallback;
}
static json_object *api(struct manager *m, const char *path, const char *method)
{
CURL *curl = curl_easy_init();
if (!curl) return NULL;
char *url = xprintf("%s%s", m->cfg.jellyfin_url, path);
char *token = xprintf("X-Emby-Token: %s", m->cfg.api_key);
struct curl_slist *headers = NULL;
headers = curl_slist_append(headers, token);
headers = curl_slist_append(headers, "Content-Length: 0");
struct buffer body = {0};
curl_easy_setopt(curl, CURLOPT_URL, url);
curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, curl_write);
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &body);
curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, 5L);
curl_easy_setopt(curl, CURLOPT_TIMEOUT, 10L);
curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L);
if (strcmp(method, "GET") != 0) curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, method);
CURLcode result = curl_easy_perform(curl);
long status = 0;
curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &status);
json_object *value = NULL;
if (result == CURLE_OK && status >= 200 && status < 300)
value = body.n ? json_tokener_parse(body.v) : json_object_new_object();
free(body.v); curl_slist_free_all(headers); curl_easy_cleanup(curl);
free(token); free(url); return value;
}
static void hash_to_hex(XXH128_hash_t hash, char out[HASH_HEX + 1])
{
XXH128_canonical_t canonical;
XXH128_canonicalFromHash(&canonical, hash);
for (size_t i = 0; i < sizeof(canonical.digest); i++)
snprintf(out + i * 2, 3, "%02x", canonical.digest[i]);
out[HASH_HEX] = 0;
}
static void hash_bytes(const void *data, size_t n, char out[HASH_HEX + 1])
{
hash_to_hex(XXH3_128bits(data, n), out);
}
static void path_hash(const char *relative, char out[HASH_HEX + 1])
{
hash_bytes(relative, strlen(relative), out);
}
static void generation_hash(const struct stat *st, char out[HASH_HEX + 1])
{
char generation[192];
int n = snprintf(generation, sizeof(generation), "%llu:%llu:%lld:%lld.%09ld",
(unsigned long long)st->st_dev, (unsigned long long)st->st_ino,
(long long)st->st_size, (long long)st->st_mtim.tv_sec, st->st_mtim.tv_nsec);
if (n < 0 || (size_t)n >= sizeof(generation)) die("source identity overflow");
hash_bytes(generation, (size_t)n, out);
}
static int file_hash(const char *path, char out[HASH_HEX + 1])
{
int fd = open(path, O_RDONLY | O_CLOEXEC | O_NOFOLLOW);
if (fd < 0) return -1;
void *data = xmalloc(HASH_CHUNK);
XXH3_state_t *state = XXH3_createState();
if (!state || XXH3_128bits_reset(state) == XXH_ERROR) die("XXH3 initialization failed");
int rc = 0;
while (!stopping) {
ssize_t n = read(fd, data, HASH_CHUNK);
if (!n) break;
if (n < 0) { if (errno == EINTR) continue; rc = -1; break; }
if (XXH3_128bits_update(state, data, (size_t)n) == XXH_ERROR) die("XXH3 failed");
}
if (stopping) { errno = EINTR; rc = -1; }
if (!rc) hash_to_hex(XXH3_128bits_digest(state), out);
int saved = errno;
XXH3_freeState(state); free(data); close(fd); errno = saved; return rc;
}
static void background_or_exit(void)
{
struct sched_param parameter = {0};
if (sched_setscheduler(0, SCHED_IDLE, &parameter) < 0) {
dprintf(STDERR_FILENO, "pretranscode: SCHED_IDLE: %s\n", strerror(errno));
_exit(126);
}
if (syscall(SYS_ioprio_set, IOPRIO_WHO_PROCESS, 0,
IOPRIO_PRIO_VALUE(IOPRIO_CLASS_IDLE, 0)) < 0) {
dprintf(STDERR_FILENO, "pretranscode: IOPRIO_CLASS_IDLE: %s\n", strerror(errno));
_exit(126);
}
}
static int background_file_hash(const char *path, char out[HASH_HEX + 1])
{
int pipes[2];
if (pipe2(pipes, O_CLOEXEC) < 0) return -1;
pid_t pid = fork();
if (pid < 0) { close(pipes[0]); close(pipes[1]); return -1; }
if (!pid) {
close(pipes[0]);
if (setsid() < 0) _exit(126);
background_or_exit();
char hash[HASH_HEX + 1];
if (file_hash(path, hash) < 0 || write(pipes[1], hash, sizeof(hash)) != (ssize_t)sizeof(hash))
_exit(1);
_exit(0);
}
close(pipes[1]);
size_t received = 0;
while (received < HASH_HEX + 1 && !stopping) {
ssize_t n = read(pipes[0], out + received, HASH_HEX + 1 - received);
if (n > 0) received += (size_t)n;
else if (!n) break;
else if (errno != EINTR) break;
}
if (stopping) kill(-pid, SIGKILL);
int status = 0;
waitpid(pid, &status, 0);
close(pipes[0]);
return received == HASH_HEX + 1 && WIFEXITED(status) && !WEXITSTATUS(status) ? 0 : -1;
}
static int capture(char *const argv[], unsigned timeout, struct buffer *output)
{
int pipes[2];
if (pipe2(pipes, O_CLOEXEC) < 0) return -1;
int flags = fcntl(pipes[0], F_GETFL);
if (flags < 0 || fcntl(pipes[0], F_SETFL, flags | O_NONBLOCK) < 0) {
close(pipes[0]); close(pipes[1]); return -1;
}
pid_t pid = fork();
if (pid < 0) { close(pipes[0]); close(pipes[1]); return -1; }
if (!pid) {
if (setsid() < 0) _exit(126);
if (dup2(pipes[1], STDOUT_FILENO) < 0 || dup2(pipes[1], STDERR_FILENO) < 0) _exit(126);
close(pipes[0]); close(pipes[1]);
background_or_exit(); execv(argv[0], argv); _exit(127);
}
close(pipes[1]);
int status = 0; bool exited = false, eof = false;
double deadline = monotonic() + timeout;
while (!exited || !eof) {
struct pollfd pfd = {.fd = pipes[0], .events = POLLIN | POLLHUP};
(void)poll(&pfd, 1, 100);
for (;;) {
char data[8192]; ssize_t n = read(pipes[0], data, sizeof(data));
if (n > 0) { buffer_add(output, data, (size_t)n); continue; }
if (!n) eof = true;
break;
}
if (waitpid(pid, &status, WNOHANG) == pid) exited = true;
if ((stopping || monotonic() >= deadline) && !exited) {
kill(-pid, SIGKILL); waitpid(pid, &status, 0); exited = true;
}
}
close(pipes[0]);
return WIFEXITED(status) ? WEXITSTATUS(status) : -1;
}
static int probe(struct manager *m, const char *path, struct media *media)
{
char *const args[] = {
m->cfg.ffprobe, "-v", "quiet", "-show_entries",
"stream=codec_type,codec_name,bits_per_raw_sample,pix_fmt,profile,color_transfer,width,height,channels:stream_side_data=side_data_type",
"-show_entries", "format=duration", "-show_chapters", "-of", "json", (char *)path, NULL
};
struct buffer output = {0};
int rc = capture(args, 120, &output);
if (rc) { free(output.v); return -1; }
json_object *root = json_tokener_parse(output.v ? output.v : "");
free(output.v); if (!root) return -1;
memset(media, 0, sizeof(*media)); media->depth = 8; media->audio_compatible = true;
json_object *streams = NULL;
bool found_video = false;
if (json_object_object_get_ex(root, "streams", &streams) && json_object_is_type(streams, json_type_array)) {
size_t count = json_object_array_length(streams);
for (size_t i = 0; i < count; i++) {
json_object *stream = json_object_array_get_idx(streams, i);
const char *type = jstring(stream, "codec_type", "");
if (!found_video && strcmp(type, "video") == 0) {
found_video = true;
snprintf(media->codec, sizeof(media->codec), "%s", jstring(stream, "codec_name", ""));
snprintf(media->profile, sizeof(media->profile), "%s", jstring(stream, "profile", ""));
snprintf(media->transfer, sizeof(media->transfer), "%s", jstring(stream, "color_transfer", ""));
media->width = jint(stream, "width", 0); media->height = jint(stream, "height", 0);
int depth = jint(stream, "bits_per_raw_sample", 0);
const char *pixel = jstring(stream, "pix_fmt", "");
if (!depth) depth = strstr(pixel, "12") || strstr(media->profile, "12") ? 12
: strstr(pixel, "10") || strstr(media->profile, "10") ? 10 : 8;
media->depth = depth;
json_object *side_data = NULL;
if (json_object_object_get_ex(stream, "side_data_list", &side_data) &&
json_object_is_type(side_data, json_type_array)) {
for (size_t j = 0; j < json_object_array_length(side_data); j++) {
json_object *side = json_object_array_get_idx(side_data, j);
if (strcasestr(jstring(side, "side_data_type", ""), "dovi"))
media->dolby_vision = true;
}
}
} else if (strcmp(type, "audio") == 0) {
media->audio++;
if (strcasecmp(jstring(stream, "codec_name", ""), "aac") || jint(stream, "channels", 0) > 2)
media->audio_compatible = false;
} else if (strcmp(type, "subtitle") == 0) media->subtitles++;
else if (strcmp(type, "attachment") == 0) media->attachments++;
}
}
json_object *format = NULL, *chapters = NULL;
if (json_object_object_get_ex(root, "format", &format)) media->duration = jdouble(format, "duration", 0);
if (json_object_object_get_ex(root, "chapters", &chapters) && json_object_is_type(chapters, json_type_array))
media->chapters = (int)json_object_array_length(chapters);
json_object_put(root); return found_video ? 0 : -1;
}
static bool incompatible(const struct media *m)
{
return strcasecmp(m->codec, "h264") || m->depth > 8 || strstr(m->profile, "10") ||
!m->audio_compatible || m->subtitles;
}
static bool hdr(const struct media *m)
{
return !strcasecmp(m->transfer, "smpte2084") || !strcasecmp(m->transfer, "arib-std-b67");
}
static char *relative_from_api(struct manager *m, const char *path)
{
if (!path) return NULL;
const char *roots[] = {m->cfg.library_root, m->cfg.source_root};
for (size_t i = 0; i < ARRAY_LEN(roots); i++) {
size_t n = strlen(roots[i]);
if (!strncmp(path, roots[i], n) && path[n] == '/' && relative_path(path + n + 1))
return xstrdup(path + n + 1);
}
return NULL;
}
static int catalog_cmp(const void *a, const void *b)
{
return strcmp(((const struct catalog_item *)a)->relative,
((const struct catalog_item *)b)->relative);
}
static void catalog_add(struct catalog *c, struct catalog_item item)
{
if (c->n == c->cap) { c->cap = c->cap ? c->cap * 2 : 256; c->v = xrealloc(c->v, c->cap * sizeof(*c->v)); }
c->v[c->n++] = item;
}
static void catalog_free(struct catalog *c)
{
for (size_t i = 0; i < c->n; i++) { free(c->v[i].relative); free(c->v[i].id); }
free(c->v); memset(c, 0, sizeof(*c));
}
static struct catalog_item *catalog_find(struct catalog *c, const char *relative)
{
struct catalog_item key = {.relative = (char *)relative};
return bsearch(&key, c->v, c->n, sizeof(*c->v), catalog_cmp);
}
static bool read_catalog(struct manager *m)
{
struct catalog next = {0};
for (int start = 0;;) {
char *path = xprintf("/Items?Recursive=true&IncludeItemTypes=Movie%%2CEpisode&Fields=Path%%2CMediaStreams&StartIndex=%d&Limit=500", start);
json_object *page = api(m, path, "GET"); free(path);
if (!page) { catalog_free(&next); return false; }
json_object *array = NULL;
if (!json_object_object_get_ex(page, "Items", &array) || !json_object_is_type(array, json_type_array)) {
json_object_put(page); catalog_free(&next); return false;
}
size_t count = json_object_array_length(array);
for (size_t i = 0; i < count; i++) {
json_object *value = json_object_array_get_idx(array, i);
char *relative = relative_from_api(m, jstring(value, "Path", NULL));
if (!relative) continue;
struct catalog_item item = {.relative = relative, .id = xstrdup(jstring(value, "Id", ""))};
item.media.depth = 8; item.media.audio_compatible = true;
json_object *streams = NULL; bool video = false;
if (json_object_object_get_ex(value, "MediaStreams", &streams) && json_object_is_type(streams, json_type_array)) {
for (size_t j = 0; j < json_object_array_length(streams); j++) {
json_object *stream = json_object_array_get_idx(streams, j);
const char *type = jstring(stream, "Type", "");
if (!video && !strcasecmp(type, "Video")) {
video = true;
snprintf(item.media.codec, sizeof(item.media.codec), "%s", jstring(stream, "Codec", ""));
snprintf(item.media.profile, sizeof(item.media.profile), "%s", jstring(stream, "Profile", ""));
snprintf(item.media.transfer, sizeof(item.media.transfer), "%s", jstring(stream, "ColorTransfer", ""));
item.media.depth = jint(stream, "BitDepth", 8);
item.media.width = jint(stream, "Width", 0); item.media.height = jint(stream, "Height", 0);
} else if (!strcasecmp(type, "Audio")) {
item.media.audio++;
if (strcasecmp(jstring(stream, "Codec", ""), "aac") || jint(stream, "Channels", 0) > 2)
item.media.audio_compatible = false;
} else if (!strcasecmp(type, "Subtitle")) item.media.subtitles++;
else if (!strcasecmp(type, "Attachment")) item.media.attachments++;
}
}
item.media.duration = jdouble(value, "RunTimeTicks", 0) / 10000000.0;
catalog_add(&next, item);
}
int total = jint(page, "TotalRecordCount", start + (int)count);
json_object_put(page); start += (int)count;
if (!count || start >= total) break;
}
qsort(next.v, next.n, sizeof(*next.v), catalog_cmp);
catalog_free(&m->catalog); m->catalog = next; return true;
}
static bool refresh(struct manager *m, const struct item *item)
{
if (!item || !item->catalog || !*item->catalog->id) return true;
char *path = xprintf("/Items/%s/Refresh?metadataRefreshMode=FullRefresh&imageRefreshMode=None&replaceAllMetadata=false&replaceAllImages=false", item->catalog->id);
json_object *reply = api(m, path, "POST"); free(path);
if (!reply) { logmsg("Unable to refresh Jellyfin item %s", item->relative); return false; }
json_object_put(reply); return true;
}
static bool parse_cache_name(const char *name, struct cache *cache)
{
size_t n = strlen(name);
const char *extension = n > 4 ? name + n - 4 : "";
if (strcmp(extension, ".mkv") && strcmp(extension, ".mp4")) return false;
size_t stem_n = n - 4;
if (stem_n != HASH_HEX * 4 + 3) return false;
char stem[HASH_HEX * 4 + 4];
memcpy(stem, name, stem_n); stem[stem_n] = 0;
char *parts[4], *save = NULL, *part = strtok_r(stem, ".", &save);
for (size_t i = 0; i < 4; i++) {
if (!part) return false;
parts[i] = part; part = strtok_r(NULL, ".", &save);
}
if (part || !hex_hash(parts[0]) || !hex_hash(parts[1]) || !hex_hash(parts[2]) || !hex_hash(parts[3]))
return false;
memcpy(cache->path_hash, parts[0], HASH_HEX + 1);
memcpy(cache->generation_hash, parts[1], HASH_HEX + 1);
memcpy(cache->profile_hash, parts[2], HASH_HEX + 1);
memcpy(cache->output_hash, parts[3], HASH_HEX + 1);
memcpy(cache->extension, extension, 5);
return true;
}
static void cache_add(struct caches *c, struct cache value)
{
if (c->n == c->cap) { c->cap = c->cap ? c->cap * 2 : 64; c->v = xrealloc(c->v, c->cap * sizeof(*c->v)); }
c->v[c->n++] = value;
}
static void caches_free(struct caches *c)
{
for (size_t i = 0; i < c->n; i++) { free(c->v[i].name); free(c->v[i].path); free(c->v[i].target); }
free(c->v); memset(c, 0, sizeof(*c));
}
static void scan_cache(struct manager *m)
{
caches_free(&m->caches);
char *root = xprintf("%s/files", m->cfg.cache_root);
if (mkdirs(root) < 0) die("mkdir %s: %s", root, strerror(errno));
DIR *directory = opendir(root);
if (!directory) die("opendir %s: %s", root, strerror(errno));
bool removed = false; struct dirent *entry;
while ((entry = readdir(directory))) {
if (entry->d_name[0] == '.' || !strcmp(entry->d_name, "lost+found")) continue;
char *path = xprintf("%s/%s", root, entry->d_name);
struct stat st;
if (lstat(path, &st) < 0 || !S_ISREG(st.st_mode)) { free(path); continue; }
if (ends(entry->d_name, ".partial")) { if (unlink(path) == 0) removed = true; free(path); continue; }
struct cache cache = {
.name = xstrdup(entry->d_name), .path = path,
.target = xprintf("%s/files/%s", m->cfg.jellyfin_cache_root, entry->d_name),
.size = (uint64_t)st.st_size,
.used_at = (double)st.st_atim.tv_sec + (double)st.st_atim.tv_nsec / 1e9,
};
cache.named = parse_cache_name(entry->d_name, &cache);
cache_add(&m->caches, cache);
}
closedir(directory); if (removed) sync_directory(root); free(root);
}
static struct cache *cache_by_target(struct manager *m, const char *target)
{
if (!target) return NULL;
for (size_t i = 0; i < m->caches.n; i++)
if (!strcmp(m->caches.v[i].target, target)) return &m->caches.v[i];
return NULL;
}
static bool cache_matches(const struct cache *cache, const struct item *item)
{
return cache && cache->named && !strcmp(cache->path_hash, item->path_hash) &&
!strcmp(cache->generation_hash, item->generation_hash) &&
!strcmp(cache->extension, item->extension);
}
static struct cache *valid_cache(struct manager *m, const struct item *item)
{
struct cache *selected = NULL;
for (size_t i = 0; i < m->caches.n; i++) {
struct cache *cache = &m->caches.v[i];
if (!cache_matches(cache, item) || strcmp(cache->profile_hash, m->cfg.profile_hash)) continue;
if (item->live_target && !strcmp(cache->target, item->live_target)) return cache;
if (!selected || cache->used_at > selected->used_at) selected = cache;
}
return selected;
}
static void item_add(struct items *items, struct item item)
{
if (items->n == items->cap) { items->cap = items->cap ? items->cap * 2 : 256; items->v = xrealloc(items->v, items->cap * sizeof(*items->v)); }
items->v[items->n++] = item;
}
static void items_free(struct items *items)
{
for (size_t i = 0; i < items->n; i++) {
free(items->v[i].relative); free(items->v[i].source); free(items->v[i].view); free(items->v[i].live_target);
}
free(items->v); memset(items, 0, sizeof(*items));
}
static struct item *item_by_path_hash(struct manager *m, const char *hash)
{
for (size_t i = 0; i < m->items.n; i++)
if (!strcmp(m->items.v[i].path_hash, hash)) return &m->items.v[i];
return NULL;
}
static double timer_get(const struct timers *timers, const char *key)
{
for (size_t i = 0; i < timers->n; i++) if (!strcmp(timers->v[i].key, key)) return timers->v[i].value;
return 0;
}
static void timer_set(struct timers *timers, const char *key, double value)
{
for (size_t i = 0; i < timers->n; i++) if (!strcmp(timers->v[i].key, key)) { timers->v[i].value = value; return; }
if (timers->n == timers->cap) { timers->cap = timers->cap ? timers->cap * 2 : 32; timers->v = xrealloc(timers->v, timers->cap * sizeof(*timers->v)); }
timers->v[timers->n++] = (struct timer){xstrdup(key), value};
}
static void timer_del(struct timers *timers, const char *key)
{
for (size_t i = 0; i < timers->n; i++) if (!strcmp(timers->v[i].key, key)) {
free(timers->v[i].key); timers->v[i] = timers->v[--timers->n]; return;
}
}
static void timers_prune(struct timers *timers, double before)
{
for (size_t i = 0; i < timers->n;)
if (timers->v[i].value <= before) {
free(timers->v[i].key); timers->v[i] = timers->v[--timers->n];
} else i++;
}
static void timers_free(struct timers *timers)
{
for (size_t i = 0; i < timers->n; i++) free(timers->v[i].key);
free(timers->v); memset(timers, 0, sizeof(*timers));
}
static char *lock_key(const char *relative)
{
if (!strncasecmp(relative, "tv/", 3)) {
const char *slash = strchr(relative + 3, '/');
if (slash) return xprintf("series:%.*s", (int)(slash - relative - 3), relative + 3);
}
return xprintf("item:%s", relative);
}
static bool locked(struct manager *m, const char *relative)
{
char *key = lock_key(relative);
bool value = strings_has(&m->active_keys, key) || timer_get(&m->cooldowns, key) > now();
free(key); return value;
}
static void touch_target(struct manager *m, const char *relative)
{
char *view = xprintf("%s/%s", m->cfg.view_root, relative);
char *target = readlink_alloc(view); free(view);
struct cache *cache = cache_by_target(m, target);
free(target);
if (!cache) return;
struct timespec times[2] = {{.tv_nsec = UTIME_NOW}, {.tv_nsec = UTIME_OMIT}};
if (utimensat(AT_FDCWD, cache->path, times, 0) == 0) cache->used_at = now();
}
static bool read_sessions(struct manager *m)
{
json_object *sessions = api(m, "/Sessions", "GET");
if (!sessions || !json_object_is_type(sessions, json_type_array)) {
if (sessions) json_object_put(sessions);
if (m->sessions_healthy) logmsg("Jellyfin sessions unavailable; pausing cache work");
m->sessions_healthy = false; strings_free(&m->active_items); strings_free(&m->active_keys); return false;
}
struct strings items = {0}, keys = {0};
for (size_t i = 0; i < json_object_array_length(sessions); i++) {
json_object *session = json_object_array_get_idx(sessions, i), *playing = NULL;
if (!json_object_object_get_ex(session, "NowPlayingItem", &playing)) continue;
char *relative = relative_from_api(m, jstring(playing, "Path", NULL));
if (!relative) {
const char *id = jstring(playing, "Id", "");
if (*id) {
char *path = xprintf("/Items/%s?Fields=Path", id);
json_object *detail = api(m, path, "GET");
free(path);
if (detail) {
relative = relative_from_api(m, jstring(detail, "Path", NULL));
json_object_put(detail);
}
}
}
if (!relative) continue;
strings_add(&items, relative); char *key = lock_key(relative); strings_add(&keys, key); free(key);
touch_target(m, relative); free(relative);
}
json_object_put(sessions); strings_sort(&items); strings_sort(&keys);
double time = now();
for (size_t i = 0; i < items.n; i++) timer_set(&m->priorities, items.v[i], time);
for (size_t i = 0; i < m->previous_items.n; i++)
if (!strings_has(&items, m->previous_items.v[i])) timer_set(&m->priorities, m->previous_items.v[i], time);
for (size_t i = 0; i < keys.n; i++) if (!strncmp(keys.v[i], "series:", 7))
timer_set(&m->cooldowns, keys.v[i], time + m->cfg.cooldown_seconds);
for (size_t i = 0; i < m->previous_keys.n; i++)
if (!strings_has(&keys, m->previous_keys.v[i]) && !strncmp(m->previous_keys.v[i], "series:", 7))
timer_set(&m->cooldowns, m->previous_keys.v[i], time + m->cfg.cooldown_seconds);
strings_free(&m->previous_items); strings_free(&m->previous_keys);
for (size_t i = 0; i < items.n; i++) strings_add(&m->previous_items, items.v[i]);
for (size_t i = 0; i < keys.n; i++) strings_add(&m->previous_keys, keys.v[i]);
strings_sort(&m->previous_items); strings_sort(&m->previous_keys);
strings_move(&m->active_items, &items); strings_move(&m->active_keys, &keys);
m->sessions_healthy = true; return true;
}
static bool cache_metadata_matches(const struct catalog_item *item)
{
return item && !strcasecmp(item->media.codec, "h264") && item->media.depth <= 8 &&
item->media.audio <= 1 && item->media.audio_compatible && !item->media.subtitles;
}
static bool source_entry(FTSENT *entry)
{
if (entry->fts_info == FTS_F) return true;
if (entry->fts_info != FTS_SL) return false;
struct stat st; return stat(entry->fts_path, &st) == 0 && S_ISREG(st.st_mode);
}
static bool mark_references(struct manager *m)
{
for (size_t i = 0; i < m->caches.n; i++) m->caches.v[i].referenced = false;
char *roots[] = {m->cfg.view_root, NULL};
FTS *tree = fts_open(roots, FTS_PHYSICAL | FTS_NOCHDIR, NULL);
if (!tree) return false;
FTSENT *entry;
while ((entry = fts_read(tree))) if (entry->fts_info == FTS_SL || entry->fts_info == FTS_SLNONE) {
char *target = readlink_alloc(entry->fts_path); struct cache *cache = cache_by_target(m, target);
if (cache)
cache->referenced = true;
free(target);
}
fts_close(tree); return true;
}
static bool reconcile(struct manager *m)
{
if (!read_catalog(m)) { logmsg("Unable to read Jellyfin inventory"); return false; }
scan_cache(m); items_free(&m->items);
struct strings seen = {0};
char *roots[] = {m->cfg.source_root, NULL};
FTS *tree = fts_open(roots, FTS_PHYSICAL | FTS_NOCHDIR, NULL);
if (!tree) { strings_free(&seen); return false; }
size_t root_n = strlen(m->cfg.source_root); FTSENT *entry; bool view_changed = false;
while ((entry = fts_read(tree))) {
const char *relative = entry->fts_path + root_n; if (*relative == '/') relative++;
if (entry->fts_info == FTS_D) {
char *path = *relative ? xprintf("%s/%s", m->cfg.view_root, relative) : xstrdup(m->cfg.view_root);
if (mkdirs(path) < 0) { free(path); fts_close(tree); strings_free(&seen); return false; }
free(path); continue;
}
if (!source_entry(entry)) continue;
strings_add(&seen, relative);
char *view = xprintf("%s/%s", m->cfg.view_root, relative);
if (!video_name(entry->fts_name)) {
char *target = xprintf("%s/%s", m->cfg.original_root, relative);
int changed = atomic_link(view, target);
if (changed < 0) logmsg("Unable to link %s: %s", relative, strerror(errno));
else if (changed > 0) view_changed = true;
free(target); free(view); continue;
}
struct stat st;
if (stat(entry->fts_path, &st) < 0 || !S_ISREG(st.st_mode)) { free(view); continue; }
struct item item = {
.relative = xstrdup(relative), .source = xstrdup(entry->fts_path), .view = view,
.extension = cache_extension(relative), .size = (uint64_t)st.st_size,
.mtime = (double)st.st_mtim.tv_sec + (double)st.st_mtim.tv_nsec / 1e9,
.catalog = catalog_find(&m->catalog, relative), .live_target = readlink_alloc(view),
};
path_hash(relative, item.path_hash); generation_hash(&st, item.generation_hash);
item.cache = valid_cache(m, &item);
struct cache *live_cache = cache_by_target(m, item.live_target);
if (live_cache != item.cache && (cache_matches(live_cache, &item) || (live_cache && !live_cache->named)))
item.fallback = live_cache;
bool is_locked = locked(m, relative);
if (item.cache) item.cache->retain = is_locked;
const char *target;
char *original = NULL;
if (is_locked && item.live_target) target = item.live_target;
else if (item.cache) target = item.cache->target;
else if (item.fallback) target = item.fallback->target;
else { original = xprintf("%s/%s", m->cfg.original_root, relative); target = original; }
bool had_live_target = item.live_target != NULL;
bool publishing_cache = item.cache && strcmp(target, item.cache->target) == 0;
int changed = atomic_link(view, target);
if (changed < 0) {
logmsg("Unable to publish %s: %s", relative, strerror(errno));
if (item.cache) item.cache->retain = true;
} else if (changed > 0) view_changed = true;
item_add(&m->items, item);
struct item *stored = &m->items.v[m->items.n - 1];
if (changed > 0) {
free(stored->live_target);
stored->live_target = xstrdup(target);
if (had_live_target || publishing_cache)
(void)refresh(m, stored);
}
else if (item.cache && !is_locked && !cache_metadata_matches(item.catalog)) (void)refresh(m, stored);
free(original);
}
fts_close(tree); strings_sort(&seen);
char *view_roots[] = {m->cfg.view_root, NULL};
tree = fts_open(view_roots, FTS_PHYSICAL | FTS_NOCHDIR, NULL);
if (tree) {
size_t view_n = strlen(m->cfg.view_root);
while ((entry = fts_read(tree))) {
const char *relative = entry->fts_path + view_n; if (*relative == '/') relative++;
if ((entry->fts_info == FTS_SL || entry->fts_info == FTS_SLNONE) && *relative &&
!strings_has(&seen, relative) && !locked(m, relative)) {
if (unlink(entry->fts_path) == 0) {
sync_parent(entry->fts_path);
view_changed = true;
}
} else if (entry->fts_info == FTS_DP && *relative) (void)rmdir(entry->fts_path);
}
fts_close(tree);
}
strings_free(&seen);
if (view_changed) {
json_object *reply = api(m, "/Library/Refresh", "POST");
if (reply) json_object_put(reply);
}
mark_references(m);
for (size_t i = 0; i < m->caches.n; i++) {
struct cache *cache = &m->caches.v[i];
if (cache->referenced || cache->retain) continue;
struct item *item = *cache->path_hash ? item_by_path_hash(m, cache->path_hash) : NULL;
if (item && !cache_by_target(m, item->live_target) && !refresh(m, item)) {
cache->retain = true;
continue;
}
if (unlink(cache->path) == 0) sync_parent(cache->path);
cache->size = 0;
}
m->last_scan = now(); return true;
}
static uint64_t cache_used(struct manager *m)
{
uint64_t used = 0;
for (size_t i = 0; i < m->caches.n; i++) used += m->caches.v[i].size;
return used;
}
static uint64_t cache_free(struct manager *m)
{
struct statvfs fs;
if (statvfs(m->cfg.cache_root, &fs) < 0) return 0;
uint64_t physical = (uint64_t)fs.f_bavail * fs.f_frsize, used = cache_used(m);
uint64_t configured = used < m->cfg.cache_max ? m->cfg.cache_max - used : 0;
return physical < configured ? physical : configured;
}
static int lru_cmp(const void *a, const void *b)
{
const struct cache *const *x = a, *const *y = b;
return (*x)->used_at < (*y)->used_at ? -1 : (*x)->used_at > (*y)->used_at ? 1 : strcmp((*x)->name, (*y)->name);
}
static bool evict(struct manager *m, uint64_t required, uint64_t target_used)
{
struct cache **order = xmalloc(m->caches.n * sizeof(*order)); size_t n = 0;
for (size_t i = 0; i < m->caches.n; i++) if (m->caches.v[i].size) order[n++] = &m->caches.v[i];
qsort(order, n, sizeof(*order), lru_cmp);
for (size_t i = 0; i < n && (cache_free(m) < required || cache_used(m) > target_used); i++) {
struct cache *cache = order[i];
struct item *item = *cache->path_hash ? item_by_path_hash(m, cache->path_hash) : NULL;
if (cache->retain || (item && locked(m, item->relative))) continue;
if (cache->referenced && item) {
char *original = xprintf("%s/%s", m->cfg.original_root, item->relative);
if (atomic_link(item->view, original) < 0) { free(original); continue; }
free(original); cache->referenced = false;
if (!refresh(m, item)) {
cache->retain = true;
continue;
}
} else if (cache->referenced) continue;
logmsg("Evicting %s", item ? item->relative : cache->name);
if (unlink(cache->path) == 0) { sync_parent(cache->path); cache->size = 0; }
}
free(order); return cache_free(m) >= required && cache_used(m) <= target_used;
}
static struct item *candidate(struct manager *m)
{
struct item *best = NULL; bool best_priority = false; double best_order = 0, time = now();
for (size_t i = 0; i < m->items.n; i++) {
struct item *item = &m->items.v[i];
bool needs_cache = item->fallback || (item->catalog && incompatible(&item->catalog->media));
if (item->cache || !needs_cache || locked(m, item->relative) ||
timer_get(&m->retries, item->relative) > time) continue;
double priority = timer_get(&m->priorities, item->relative);
bool preferred = priority > 0; double order = preferred ? priority : item->mtime;
if (!best || preferred > best_priority || (preferred == best_priority && order > best_order)) {
best = item; best_priority = preferred; best_order = order;
}
}
return best;
}
static void cancel_child(struct manager *m, const char *reason)
{
if (m->child <= 0) return;
logmsg("Stopping encode: %s", reason); kill(-m->child, SIGTERM);
for (int i = 0; i < 100; i++) {
if (waitpid(m->child, NULL, WNOHANG) == m->child) { m->child = 0; return; }
struct timespec delay = {.tv_nsec = 100000000}; nanosleep(&delay, NULL);
}
kill(-m->child, SIGKILL); waitpid(m->child, NULL, 0); m->child = 0;
}
static pid_t start_ffmpeg(struct manager *m, const struct item *item, const char *partial, const struct media *source)
{
bool is_hdr = hdr(source);
const char *filter = !is_hdr ? "zscale=dither=error_diffusion,format=yuv420p" :
!strcasecmp(source->transfer, "arib-std-b67") ?
"zscale=tin=arib-std-b67:pin=bt2020:min=bt2020nc:t=linear:npl=100,format=gbrpf32le,zscale=p=bt709,tonemap=tonemap=hable:desat=0,zscale=t=bt709:m=bt709:r=tv:dither=error_diffusion,format=yuv420p" :
"zscale=t=linear:npl=100,format=gbrpf32le,zscale=p=bt709,tonemap=tonemap=hable:desat=0,zscale=t=bt709:m=bt709:r=tv:dither=error_diffusion,format=yuv420p";
char *graph = xprintf("[0:v:0]%s[video]", filter);
char *args[64]; size_t n = 0;
#define A(x) do { if (n + 1 >= ARRAY_LEN(args)) die("ffmpeg argument overflow"); args[n++] = (char *)(x); } while (0)
A(m->cfg.ffmpeg); A("-nostdin"); A("-hide_banner"); A("-loglevel"); A("warning"); A("-i"); A(item->source);
A("-filter_complex"); A(graph); A("-map"); A("[video]"); A("-map"); A("0:a:0?");
A("-map_metadata"); A("-1"); A("-map_chapters"); A("-1");
A("-c:v"); A("libx264"); A("-preset"); A(m->cfg.preset); A("-crf"); A(m->cfg.crf);
A("-maxrate"); A(m->cfg.maxrate); A("-bufsize"); A(m->cfg.bufsize); A("-pix_fmt"); A("yuv420p"); A("-profile:v"); A("high");
A("-c:a"); A("aac"); A("-ac:a"); A("2"); A("-b:a"); A(AUDIO_BITRATE);
if (is_hdr) { A("-color_primaries"); A("bt709"); A("-color_trc"); A("bt709"); A("-colorspace"); A("bt709"); A("-color_range"); A("tv"); }
if (!strcmp(item->extension, ".mp4")) { A("-movflags"); A("+faststart"); A("-f"); A("mp4"); }
else { A("-f"); A("matroska"); }
A(partial); args[n] = NULL;
#undef A
char *log = xprintf("%s/ffmpeg.log", m->cfg.state_root);
int fd = open(log, O_WRONLY | O_CREAT | O_APPEND | O_CLOEXEC, 0644); free(log);
if (fd < 0) return -1;
pid_t pid = fork();
if (pid < 0) { close(fd); free(graph); return -1; }
if (!pid) {
if (setsid() < 0 || dup2(fd, STDOUT_FILENO) < 0 || dup2(fd, STDERR_FILENO) < 0) _exit(126);
close(fd);
background_or_exit(); execv(args[0], args); _exit(127);
}
close(fd); free(graph); return pid;
}
static bool valid_output(const struct media *source, const struct media *output)
{
double tolerance = source->duration * .01 > 3 ? source->duration * .01 : 3;
int expected_audio = source->audio ? 1 : 0;
return !strcasecmp(output->codec, "h264") && output->depth <= 8 && !output->dolby_vision &&
(!source->duration || (output->duration >= source->duration - tolerance && output->duration <= source->duration + tolerance)) &&
(!source->width || source->width == output->width) && (!source->height || source->height == output->height) &&
output->audio == expected_audio && output->audio_compatible && !output->subtitles &&
!output->attachments && !output->chapters;
}
static int publish(struct manager *m, struct item *item)
{
bool preferred = timer_get(&m->priorities, item->relative) > 0;
uint64_t estimate = (uint64_t)((long double)item->size * 2.2L);
if (estimate < 1000000000ULL) estimate = 1000000000ULL;
if (estimate > 100000000000ULL) estimate = 100000000000ULL;
if (cache_free(m) < estimate && (!preferred || !evict(m, estimate, m->cfg.cache_max))) return 2;
struct media source, output;
if (probe(m, item->source, &source) < 0) return -1;
if (!incompatible(&source)) {
char *original = xprintf("%s/%s", m->cfg.original_root, item->relative);
int changed = atomic_link(item->view, original); free(original);
if (changed < 0) return -1;
if (changed > 0) (void)refresh(m, item);
m->last_scan = 0; return changed;
}
char *root = xprintf("%s/files", m->cfg.cache_root);
char *partial = xprintf("%s/%s.%s.%s.%ld%s.partial", root, item->path_hash,
item->generation_hash, m->cfg.profile_hash, (long)getpid(), item->extension);
unlink(partial); logmsg("Encoding %s", item->relative);
m->child = start_ffmpeg(m, item, partial, &source);
if (m->child < 0) { free(root); free(partial); return -1; }
int status = 0;
while (!stopping) {
if (waitpid(m->child, &status, WNOHANG) == m->child) { m->child = 0; break; }
sleep(m->cfg.poll_seconds);
if (!read_sessions(m) || locked(m, item->relative)) {
cancel_child(m, "item or series is being watched"); unlink(partial); sync_directory(root);
free(root); free(partial); return 0;
}
}
if (stopping) { cancel_child(m, "service stopping"); unlink(partial); sync_directory(root); free(root); free(partial); return 0; }
if (!WIFEXITED(status) || WEXITSTATUS(status)) { unlink(partial); sync_directory(root); free(root); free(partial); return -1; }
struct stat st;
if (stat(partial, &st) < 0 || st.st_size < 1000000 || probe(m, partial, &output) < 0 || !valid_output(&source, &output)) {
unlink(partial); sync_directory(root); free(root); free(partial); return -1;
}
char output_hash[HASH_HEX + 1];
if (background_file_hash(partial, output_hash) < 0) { unlink(partial); sync_directory(root); free(root); free(partial); return -1; }
int fd = open(partial, O_RDONLY | O_CLOEXEC | O_NOFOLLOW);
if (fd < 0 || fchmod(fd, 0444) < 0 || fsync(fd) < 0) {
if (fd >= 0) close(fd);
unlink(partial); sync_directory(root); free(root); free(partial); return -1;
}
close(fd);
char *name = xprintf("%s.%s.%s.%s%s", item->path_hash, item->generation_hash,
m->cfg.profile_hash, output_hash, item->extension);
char *final = xprintf("%s/%s", root, name);
if (syscall(SYS_renameat2, AT_FDCWD, partial, AT_FDCWD, final, RENAME_NOREPLACE) < 0) {
unlink(partial); free(name); free(final); free(root); free(partial); return -1;
}
if (sync_directory(root) < 0) { free(name); free(final); free(root); free(partial); return -1; }
if (locked(m, item->relative)) { unlink(final); sync_directory(root); free(name); free(final); free(root); free(partial); return 0; }
char *target = xprintf("%s/files/%s", m->cfg.jellyfin_cache_root, name);
if (atomic_link(item->view, target) < 0) { free(target); free(name); free(final); free(root); free(partial); return -1; }
(void)refresh(m, item); timer_del(&m->priorities, item->relative); timer_del(&m->retries, item->relative);
logmsg("Completed %s", item->relative);
free(target); free(name); free(final); free(root); free(partial); m->last_scan = 0; return 1;
}
static void signal_handler(int signal_number) { (void)signal_number; stopping = 1; }
static uint64_t number(const char *name, const char *value)
{
char *end = NULL; errno = 0; unsigned long long n = strtoull(value, &end, 10);
if (errno || !end || *end)
die("invalid %s: %s", name, value);
return (uint64_t)n;
}
static const char *environment_value(const char *name, const char *fallback)
{
const char *value = getenv(name);
return value && *value ? value : fallback;
}
static char *environment(const char *name, const char *fallback)
{
const char *value = getenv(name); return xstrdup(value && *value ? value : fallback);
}
static void config_load(struct config *c)
{
memset(c, 0, sizeof(*c));
c->source_root = environment("SOURCE_ROOT", "/source"); c->view_root = environment("VIEW_ROOT", "/view");
c->cache_root = environment("CACHE_ROOT", "/cache"); c->state_root = environment("STATE_ROOT", "/state");
c->jellyfin_url = environment("JELLYFIN_URL", "http://jellyfin:8096"); c->library_root = environment("JELLYFIN_LIBRARY_ROOT", "/data/library");
c->original_root = environment("JELLYFIN_ORIGINAL_ROOT", "/data/original"); c->jellyfin_cache_root = environment("JELLYFIN_CACHE_ROOT", "/data/cache");
c->api_key = environment("JELLYFIN_API_KEY", ""); c->ffmpeg = environment("FFMPEG", "/usr/bin/ffmpeg"); c->ffprobe = environment("FFPROBE", "/usr/bin/ffprobe");
c->crf = environment("H264_CRF", "18"); c->preset = environment("H264_PRESET", "medium"); c->maxrate = environment("H264_MAXRATE", "15M"); c->bufsize = environment("H264_BUFSIZE", "30M");
c->profile = xprintf("h264-high-8bit-aac-stereo-%s-nosubs-crf%s-%s-%s-%s-canonical-v1",
AUDIO_BITRATE, c->crf, c->preset, c->maxrate, c->bufsize);
hash_bytes(c->profile, strlen(c->profile), c->profile_hash);
c->cache_max = number("CACHE_MAX_BYTES", environment_value("CACHE_MAX_BYTES", "1000000000000"));
c->cooldown_seconds = (unsigned)number("SERIES_COOLDOWN_SECONDS", environment_value("SERIES_COOLDOWN_SECONDS", "3600"));
c->poll_seconds = (unsigned)number("POLL_SECONDS", environment_value("POLL_SECONDS", "5"));
c->rescan_seconds = (unsigned)number("RESCAN_SECONDS", environment_value("RESCAN_SECONDS", "300"));
if (!*c->api_key || !c->poll_seconds || !c->rescan_seconds) die("API key and positive intervals are required");
}
static void config_free(struct config *c)
{
char *values[] = {c->source_root,c->view_root,c->cache_root,c->state_root,c->jellyfin_url,c->library_root,c->original_root,c->jellyfin_cache_root,c->api_key,c->ffmpeg,c->ffprobe,c->crf,c->preset,c->maxrate,c->bufsize,c->profile};
for (size_t i = 0; i < ARRAY_LEN(values); i++) free(values[i]);
}
static void manager_init(struct manager *m)
{
memset(m, 0, sizeof(*m)); config_load(&m->cfg);
if (mkdirs(m->cfg.view_root) < 0 || mkdirs(m->cfg.state_root) < 0) die("unable to create working directories");
char *marker = xprintf("%s/.pretranscode-cache", m->cfg.cache_root); struct stat st;
if (stat(marker, &st) < 0 || !S_ISREG(st.st_mode)) die("cache filesystem is not mounted at %s", m->cfg.cache_root);
free(marker);
m->cache_lock = open(m->cfg.cache_root, O_RDONLY | O_DIRECTORY | O_CLOEXEC);
if (m->cache_lock < 0 || flock(m->cache_lock, LOCK_EX | LOCK_NB) < 0) die("another controller owns the cache");
struct sigaction action = {.sa_handler = signal_handler}; sigemptyset(&action.sa_mask);
sigaction(SIGTERM, &action, NULL); sigaction(SIGINT, &action, NULL);
}
static void manager_free(struct manager *m)
{
if (m->child > 0) cancel_child(m, "service stopping");
strings_free(&m->active_items); strings_free(&m->active_keys); strings_free(&m->previous_items); strings_free(&m->previous_keys);
timers_free(&m->cooldowns); timers_free(&m->priorities); timers_free(&m->retries);
catalog_free(&m->catalog); caches_free(&m->caches); items_free(&m->items);
if (m->cache_lock >= 0)
close(m->cache_lock);
config_free(&m->cfg);
}
static void run(struct manager *m)
{
while (!stopping && !read_sessions(m)) sleep(m->cfg.poll_seconds);
if (stopping || !reconcile(m)) return;
(void)evict(m, 0, m->cfg.cache_max);
while (!stopping) {
if (!read_sessions(m)) { sleep(m->cfg.poll_seconds); continue; }
if (now() - m->last_scan >= m->cfg.rescan_seconds) {
if (!reconcile(m)) { sleep(m->cfg.poll_seconds); continue; }
(void)evict(m, 0, m->cfg.cache_max);
}
timers_prune(&m->cooldowns, now()); timers_prune(&m->retries, now());
struct item *item = candidate(m);
if (!item) { sleep(m->cfg.poll_seconds); continue; }
uint64_t estimate = (uint64_t)((long double)item->size * 2.2L);
if (estimate < 1000000000ULL) estimate = 1000000000ULL;
if (estimate > 100000000000ULL) estimate = 100000000000ULL;
bool preferred = timer_get(&m->priorities, item->relative) > 0;
if (!preferred && cache_free(m) < estimate) { sleep(60); continue; }
int result = publish(m, item);
if (result < 0) { logmsg("Encode failed for %s", item->relative); timer_set(&m->retries, item->relative, now() + 21600); }
if (!stopping && m->last_scan == 0) (void)reconcile(m);
}
}
#ifndef PRETRANSCODE_TEST
int main(int argc, char **argv)
{
if (argc == 2 && !strcmp(argv[1], "--version")) { puts("pretranscode " VERSION); return 0; }
if (curl_global_init(CURL_GLOBAL_DEFAULT) != CURLE_OK) die("curl initialization failed");
struct manager manager; manager_init(&manager);
const char *command = argc > 1 ? argv[1] : "run"; int result = EXIT_SUCCESS;
if (!strcmp(command, "run")) run(&manager);
else if (!strcmp(command, "sync")) { if (!read_sessions(&manager) || !reconcile(&manager)) result = EXIT_FAILURE; }
else if (!strcmp(command, "active")) result = !read_sessions(&manager) || manager.active_items.n ? EXIT_FAILURE : EXIT_SUCCESS;
else if (!strcmp(command, "purge")) {
if (!read_sessions(&manager) || manager.active_items.n || !reconcile(&manager)) result = EXIT_FAILURE;
else if (!evict(&manager, 0, 0)) result = EXIT_FAILURE;
} else if (!strcmp(command, "trim") && argc == 3) {
uint64_t target = number("trim size", argv[2]); target = target > 5000000000ULL ? target - 5000000000ULL : 0;
if (!read_sessions(&manager) || manager.active_items.n || !reconcile(&manager) || !evict(&manager, 0, target)) result = EXIT_FAILURE;
} else { fprintf(stderr, "usage: %s [run|sync|active|purge|trim BYTES|--version]\n", argv[0]); result = EXIT_FAILURE; }
manager_free(&manager); curl_global_cleanup(); return result;
}
#endif
Reference in New Issue View Git Blame Copy Permalink