dotnet_dbsync/MongoLoader.cs

using MongoDB.Bson;
using MongoDB.Bson.Serialization;
using MongoDB.Driver;
using Serilog;
using System.Text.Json;

public class MongoLoader
{
    private readonly IMongoCollection<BsonDocument> _series;
    private readonly IMongoCollection<BsonDocument> _episodes;
    private readonly IMongoCollection<BsonDocument> _actors;
    private readonly IMongoCollection<BsonDocument> _characters;
    private readonly IMongoCollection<BsonDocument> _crew;

    public MongoLoader(IMongoDatabase db)
    {
        _series = db.GetCollection<BsonDocument>("series");
        _episodes = db.GetCollection<BsonDocument>("episodes");
        _actors = db.GetCollection<BsonDocument>("actors");
        _characters = db.GetCollection<BsonDocument>("characters");
        _crew = db.GetCollection<BsonDocument>("crew");
    }

    public async Task<(int Success, int Failed)> LoadJsonToMongoAsync(
        Dictionary<string, string> directories,
        List<string> updateList,
        Dictionary<string, long> updateDict)
    {
        int success = 0;
        int failed = 0;

        var options = new ParallelOptions
        {
            MaxDegreeOfParallelism = Environment.ProcessorCount // or set manually (e.g. 4)
        };

        await Parallel.ForEachAsync(updateList, options, async (seriesId, cancellationToken) =>
        {
            try
            {
                var seriesFile = Path.Combine(directories["SERIESDIR"], $"{seriesId}.json");
                if (!File.Exists(seriesFile))
                    return;

                var json = await File.ReadAllTextAsync(seriesFile, cancellationToken);
                var doc = BsonDocument.Parse(json);

                await UpsertWithTimestamp(_series, doc, updateDict, seriesId);

                await ProcessEpisodes(seriesId, directories, updateDict);
                await ProcessCast(seriesId, directories, updateDict);
                await ProcessCrew(seriesId, directories, updateDict);

                Interlocked.Increment(ref success);

                Log.Information("Processed series {SeriesId}", seriesId);
            }
            catch (Exception ex)
            {
                Interlocked.Increment(ref failed);
                Log.Error(ex, "Failed processing series {SeriesId}", seriesId);
            }
        });

        return (success, failed);
    }


    private async Task UpsertWithTimestamp(
        IMongoCollection<BsonDocument> collection,
        BsonDocument doc,
        Dictionary<string, long> updateDict,
        string seriesId)
    {
        var id = doc.GetValue("id", null);
        if (id == null) return;

        var filter = Builders<BsonDocument>.Filter.Eq("id", id);
        var existing = await collection.Find(filter).FirstOrDefaultAsync();

        if (updateDict.TryGetValue(seriesId, out var newTimestamp))
        {
            if (existing == null ||
                !existing.Contains("updated") ||
                existing["updated"].AsInt64 < newTimestamp)
            {
                await collection.ReplaceOneAsync(filter, doc,
                    new ReplaceOptions { IsUpsert = true });
            }
        }
    }

    private async Task ProcessEpisodes(
        string seriesId,
        Dictionary<string, string> directories,
        Dictionary<string, long> updateDict)
    {
        var file = Path.Combine(directories["EPISODEDIR"], $"{seriesId}.json");
        if (!File.Exists(file)) return;

        var json = await File.ReadAllTextAsync(file);
        var array = JsonSerializer.Deserialize<List<Dictionary<string, object>>>(json);

        if (array == null) return;

        foreach (var episode in array)
        {
            episode["seriesid"] = int.Parse(seriesId);
            var doc = BsonDocument.Parse(JsonSerializer.Serialize(episode));
            await UpsertWithTimestamp(_episodes, doc, updateDict, seriesId);
        }
    }

    private async Task ProcessCast(
        string seriesId,
        Dictionary<string, string> directories,
        Dictionary<string, long> updateDict)
    {
        var file = Path.Combine(directories["CASTDIR"], $"{seriesId}.json");
        if (!File.Exists(file)) return;

        var json = await File.ReadAllTextAsync(file);

        // ✅ Correct way to parse root JSON array
        var array = BsonSerializer.Deserialize<BsonArray>(json);

        foreach (var item in array)
        {
            if (!item.IsBsonDocument)
                continue;

            var doc = item.AsBsonDocument;

            if (doc.Contains("person") && doc["person"].IsBsonDocument)
            {
                var person = doc["person"].AsBsonDocument;
                person["seriesid"] = int.Parse(seriesId);

                await UpsertWithTimestamp(_actors, person, updateDict, seriesId);
            }

            if (doc.Contains("character") && doc["character"].IsBsonDocument)
            {
                var character = doc["character"].AsBsonDocument;
                character["seriesid"] = int.Parse(seriesId);

                await UpsertWithTimestamp(_characters, character, updateDict, seriesId);
            }
        }
    }

    private async Task ProcessCrew(
        string seriesId,
        Dictionary<string, string> directories,
        Dictionary<string, long> updateDict)
    {
        var file = Path.Combine(directories["CREWDIR"], $"{seriesId}.json");
        if (!File.Exists(file)) return;

        var json = await File.ReadAllTextAsync(file);

        // ✅ Correct way to parse root JSON array
        var array = BsonSerializer.Deserialize<BsonArray>(json);

        foreach (var item in array)
        {
            if (!item.IsBsonDocument)
                continue;

            var doc = item.AsBsonDocument;

            if (doc.Contains("person") && doc["person"].IsBsonDocument)
            {
                var person = doc["person"].AsBsonDocument;

                person["seriesid"] = int.Parse(seriesId);

                // Safely get crew type
                person["crew_type"] = doc.Contains("type")
                    ? doc["type"]
                    : "unknown";

                await UpsertWithTimestamp(_crew, person, updateDict, seriesId);
            }
        }
    }
}