Repository: microsoft/onefuzz
File: src/ApiService/ApiService/Functions/QueueFileChanges.cs (C#)

Code before:
        const string eventType = "eventType";
        if (!fileChangeEvent.RootElement.TryGetProperty(eventType, out var eventTypeElement)
            || eventTypeElement.GetString() != "Microsoft.Storage.BlobCreated") {
            return;
        }

        const string topic = "topic";
        if (!fileChangeEvent.RootElement.TryGetProperty(topic, out var topicElement)
            || !_storage.CorpusAccounts().Contains(new ResourceIdentifier(topicElement.GetString()!))) {
            return;
        }

        await FileAdded(_log, fileChangeEvent, lastTry);
    }

    [Function("QueueFileChangesPoison")]
    public async Async.Task Poison(
        [QueueTrigger(QueueFileChangesPoisonQueueName, Connection = "AzureWebJobsStorage")] string msg) {
        var fileChangeEvent = JsonSerializer.Deserialize<JsonDocument>(msg, EntityConverter.GetJsonSerializerOptions());

        _ = fileChangeEvent ?? throw new ArgumentException("Unable to parse queue trigger as JSON");

        // check type first before calling Azure APIs
        const string eventType = "eventType";
        if (!fileChangeEvent.RootElement.TryGetProperty(eventType, out var eventTypeElement)
            || eventTypeElement.GetString() != "Microsoft.Storage.BlobCreated") {
            return;
        }

        const string topic = "topic";
        if (!fileChangeEvent.RootElement.TryGetProperty(topic, out var topicElement)
            || !_storage.CorpusAccounts().Contains(new ResourceIdentifier(topicElement.GetString()!))) {
            return;
        }

        try {
            // Setting isLastRetryAttempt to false will rethrow any exceptions
            // With the intention that the azure functions runtime will handle requeing
            // the message for us. The difference is for the poison queue, we're handling the
            // requeuing ourselves because azure functions doesn't support retry policies
            // for queue based functions.
            await FileAdded(_log, fileChangeEvent, false);
        } catch {
            await RequeueMessage(msg);
        }
    }

    private async Async.Task FileAdded(ILogTracer log, JsonDocument fileChangeEvent, bool isLastRetryAttempt) {
        var data = fileChangeEvent.RootElement.GetProperty("data");
        var url = data.GetProperty("url").GetString()!;
        var parts = url.Split("/").Skip(3).ToList();

Review comment (suggestion):
I think we can add this logic to the existing `QueueFileChanges` function and keep the poison queue for message that need manual intervention. we will avoid splitting the processing code in two places.

Code after:
        const string expectedEventType = "Microsoft.Storage.BlobCreated";
        if (!fileChangeEvent.RootElement.TryGetProperty(eventType, out var eventTypeElement)
            || eventTypeElement.GetString() != expectedEventType) {
            _log.WithTag("queueMessage", msg)
                .Info($"Expected fileChangeEvent to contain a property named '{eventType}' with value equal to {expectedEventType}");
            return;
        }

        const string topic = "topic";
        if (!fileChangeEvent.RootElement.TryGetProperty(topic, out var topicElement)
            || !_storage.CorpusAccounts().Contains(new ResourceIdentifier(topicElement.GetString()!))) {
            _log.WithTag("queueMessage", msg)
                .Info($"Expected fileChangeEvent to contain a property named '{topic}' with a value contained in corpus accounts");
            return;
        }

        try {
            // Setting isLastRetryAttempt to false will rethrow any exceptions
            // With the intention that the azure functions runtime will handle requeing
            // the message for us. The difference is for the poison queue, we're handling the
            // requeuing ourselves because azure functions doesn't support retry policies
            // for queue based functions.

            await FileAdded(_log, fileChangeEvent, isLastRetryAttempt: false);
        } catch {
            await RequeueMessage(msg);
        }
    }

    private async Async.Task FileAdded(ILogTracer log, JsonDocument fileChangeEvent, bool isLastRetryAttempt) {
        var data = fileChangeEvent.RootElement.GetProperty("data");
        var url = data.GetProperty("url").GetString()!;
        var parts = url.Split("/").Skip(3).ToList();

        var container = parts[0];
        var path = string.Join('/', parts.Skip(1));

        log.Info($"file added : {container:Tag:Container} - {path:Tag:Path}");
        await _notificationOperations.NewFiles(Container.Parse(container), path, isLastRetryAttempt);
    }

    private async Async.Task RequeueMessage(string msg) {
        var json = JsonNode.Parse(msg);

        // Messages that are 'manually' requeued by us as opposed to being requeued by the azure functions runtime
        // are treated as new every single time. Thus, the dequeue count doesn't persist.
        // That's why we need to keep track of it ourselves.
        var newCustomDequeueCount = 0;
        if (json!["data"]?["customDequeueCount"] != null) {
            newCustomDequeueCount = json["data"]!["customDequeueCount"]!.GetValue<int>();
        }