utun2/src/media_async/media_async.c

// media_async.c — thread-per-task async engine + crypto/file primitives
#include "media_async.h"
#include "../../lib/u_async.h"
#include "../../lib/mem.h"
#include "../../lib/debug_config.h"

#include <openssl/sha.h>
#include <openssl/evp.h>
#include <openssl/rand.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>

#ifdef _WIN32
    #include <windows.h>
#else
    #include <dirent.h>
#endif

struct ma_thread_ctx;
struct media_async { struct ma_thread_ctx* jobs; int closing; };

struct ma_thread_ctx {
    struct media_async* owner;
    struct ma_thread_ctx* next;
    struct UASYNC* ua;
    pthread_t thread;
    struct posted_task* completion;
    ma_work_fn work;
    void* data;
    ma_done_fn done;
    void* arg;
};

static void ma_finish(struct ma_thread_ctx* job, int error) {
    struct ma_thread_ctx** p = &job->owner->jobs;
    while (*p && *p != job) p = &(*p)->next;
    if (*p) *p = job->next;
    pthread_join(job->thread, NULL);
    job->done(job->arg, error);
    u_free(job);
}

/* Completion приходит только после работы; ожидать незавершённый worker здесь нельзя. */
static void ma_completed(void* arg) {
    struct ma_thread_ctx* job = arg;
    job->completion = NULL;
    ma_finish(job, 0);
}

static void* ma_thread_entry(void* arg) {
    struct ma_thread_ctx* job = arg;
    job->work(job->data);
    uasync_post_reserved(job->ua, job->completion);
    return NULL;
}

struct media_async* media_async_create(void) {
    struct media_async* ma = u_calloc(1, sizeof(*ma));
    if (!ma) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media worker owner allocation failed");
    else DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "media_async: created");
    return ma;
}

void media_async_destroy(struct media_async* ma) {
    if (!ma) return;
    ma->closing = 1;
    unsigned cancelled = 0;
    while (ma->jobs) {
        struct ma_thread_ctx* job = ma->jobs;
        pthread_join(job->thread, NULL);
        uasync_cancel_post(job->ua, job->completion);
        ma->jobs = job->next;
        job->done(job->arg, MEDIA_ASYNC_CANCELLED);
        u_free(job); cancelled++;
    }
    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "media_async: destroyed cancelled=%u", cancelled);
    u_free(ma);
}

void media_async_submit(struct media_async* ma, struct UASYNC* ua,
                        ma_work_fn work, void* data, ma_done_fn done, void* arg) {
    if (!ma || !ua || !work || !done || ma->closing) {
        DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "media task rejected: invalid arguments or stopping owner");
        if (done) done(arg, MEDIA_ASYNC_CANCELLED);
        return;
    }
    unsigned active = 0;
    for (struct ma_thread_ctx* pending = ma->jobs; pending; pending = pending->next) active++;
    if (active >= 16) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "media task limit reached active=%u", active); done(arg, -1); return; }
    struct ma_thread_ctx* job = u_calloc(1, sizeof(*job));
    if (!job) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media task allocation failed"); done(arg, -1); return; }
    job->owner = ma; job->ua = ua; job->work = work; job->data = data; job->done = done; job->arg = arg;
    job->completion = u_calloc(1, sizeof(*job->completion));
    if (!job->completion) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media completion allocation failed"); u_free(job); done(arg, -1); return; }
    job->completion->callback = ma_completed;
    job->completion->arg = job;
    int rc = pthread_create(&job->thread, NULL, ma_thread_entry, job);
    if (rc) {
        DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media task thread failed rc=%d", rc);
        u_free(job->completion); u_free(job); done(arg, -1); return;
    }
    job->next = ma->jobs; ma->jobs = job;
}

/* ─── примитивы (вызываются из worker thread) ─── */

int ma_sha256_file(const char* path, uint8_t hash_out[32]) {
    FILE* f = fopen(path, "rb");
    if (!f) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "ma_sha256_file: cannot open %s", path); return -1; }

    SHA256_CTX ctx;
    SHA256_Init(&ctx);

    uint8_t buf[65536];
    size_t rd;
    while ((rd = fread(buf, 1, sizeof(buf), f)) > 0) SHA256_Update(&ctx, buf, rd);
    int error = ferror(f);
    if (fclose(f)) error = 1;
    if (error) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "ma_sha256_file: read/close failed path=%s", path); return -1; }
    SHA256_Final(hash_out, &ctx);
    return 0;
}

int ma_sign_block(const uint8_t* ed25519_privkey, const uint8_t* data, size_t len, uint8_t sig_out[64]) {
    EVP_PKEY* pkey = EVP_PKEY_new_raw_private_key(EVP_PKEY_ED25519, NULL, ed25519_privkey, 32);
    if (!pkey) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "ma_sign_block: EVP_PKEY_new_raw_private_key failed"); return -1; }

    EVP_MD_CTX* mdctx = EVP_MD_CTX_new();
    if (!mdctx) { EVP_PKEY_free(pkey); return -1; }

    if (EVP_DigestSignInit(mdctx, NULL, NULL, NULL, pkey) != 1) {
        DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "ma_sign_block: EVP_DigestSignInit failed");
        EVP_MD_CTX_free(mdctx); EVP_PKEY_free(pkey); return -1;
    }

    size_t siglen = 64;
    int rc = EVP_DigestSign(mdctx, sig_out, &siglen, data, len);
    EVP_MD_CTX_free(mdctx);
    EVP_PKEY_free(pkey);

    if (rc != 1 || siglen != 64) {
        DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "ma_sign_block: EVP_DigestSign failed rc=%d len=%zu", rc, siglen);
        return -1;
    }
    return 0;
}

int ma_copy_file(const char* src, const char* dst) {
    if (!src || !dst) return -1;
    if (strcmp(src, dst) == 0) return 0;

    FILE* s = fopen(src, "rb");
    if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "ma_copy_file: cannot open src %s", src); return -1; }

    FILE* d = fopen(dst, "wb");
    if (!d) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "ma_copy_file: cannot create dst %s", dst); fclose(s); return -1; }

    uint8_t buf[65536];
    size_t rd;
    while ((rd = fread(buf, 1, sizeof(buf), s)) > 0) {
        if (fwrite(buf, 1, rd, d) != rd) {
            DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "ma_copy_file: write error at %s", dst);
            fclose(s); fclose(d); return -1;
        }
    }
    int error = ferror(s);
    if (fclose(s)) error = 1;
    if (fclose(d)) error = 1;
    if (error) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "ma_copy_file: read/close failed src=%s dst=%s", src, dst);
    return error ? -1 : 0;
}

int ma_file_size(const char* path) {
    struct stat st;
    if (stat(path, &st) != 0) return -1;
    return (int)(st.st_size > 0 ? st.st_size : 0);
}

void ma_uuid(uint8_t uuid_out[16]) {
    RAND_bytes(uuid_out, 16);
}

/* ─── Рекурсивный обход каталога (кроссплатформенный) ─── */

struct ma_walk_ctx { ma_dir_walk_cb cb; void* arg; };

#ifndef _WIN32
static void ma_walk_recursive(const char* dir, struct ma_walk_ctx* wc) {
    DIR* d = opendir(dir);
    if (!d) return;
    struct dirent* e;
    while ((e = readdir(d)) != NULL) {
        const char* n = e->d_name;
        if (n[0] == '.' && (n[1] == '\0' || (n[1] == '.' && n[2] == '\0'))) continue;
        char full[4096];
        snprintf(full, sizeof(full), "%s/%s", dir, n);
        struct stat st;
        if (lstat(full, &st) != 0) continue;
        if (S_ISDIR(st.st_mode)) ma_walk_recursive(full, wc);
        else if (S_ISREG(st.st_mode)) wc->cb(wc->arg, full, (int64_t)st.st_size);
    }
    closedir(d);
}
#else
static void ma_walk_recursive(const char* dir, struct ma_walk_ctx* wc) {
    char pattern[4096];
    snprintf(pattern, sizeof(pattern), "%s/*", dir);
    WIN32_FIND_DATAA fd;
    HANDLE h = FindFirstFileA(pattern, &fd);
    if (h == INVALID_HANDLE_VALUE) return;
    do {
        const char* n = fd.cFileName;
        if (n[0] == '.' && (n[1] == '\0' || (n[1] == '.' && n[2] == '\0'))) continue;
        char full[4096];
        snprintf(full, sizeof(full), "%s/%s", dir, n);
        if (fd.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY) ma_walk_recursive(full, wc);
        else {
            LARGE_INTEGER sz; sz.HighPart = fd.nFileSizeHigh; sz.LowPart = fd.nFileSizeLow;
            wc->cb(wc->arg, full, (int64_t)sz.QuadPart);
        }
    } while (FindNextFileA(h, &fd));
    FindClose(h);
}
#endif

int ma_dir_walk(const char* dir, ma_dir_walk_cb cb, void* arg) {
    if (!dir || !dir[0] || !cb) return -1;
    struct ma_walk_ctx wc; wc.cb = cb; wc.arg = arg;
    ma_walk_recursive(dir, &wc);
    return 0;
}