customconnector.cs

using System;
using System.IO;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Caching.Memory;
using IOBoard.DataIngestor.Common;
using System.Linq;
using SInnovations.Azure.TableStorageRepository.Queryable;
using System.Text;
using System.Collections.Generic;

namespace IOBoard.DataIngestor.ServiceProvider
{

    [ApiController]
    public class IngestorEndpoint : ControllerBase
    {
        private readonly IMemoryCache memoryCache;

        public IngestorEndpoint(IMemoryCache memoryCache)
        {
            this.memoryCache = memoryCache ?? throw new ArgumentNullException(nameof(memoryCache));
        }

        [HttpGet("providers/IO-Board.DataIngestor/swagger/authorize")]
        public async Task<IActionResult> authorize()
        {
            var url = Request.Query["redirect_uri"] + "?state=" + Request.Query["state"] +"&code=test";

            return Redirect(url);
           // return Ok("hello world");
        }

        [HttpPost("providers/IO-Board.DataIngestor/swagger/token")]
        public async Task<IActionResult> token()
        {
            var form = await Request.ReadFormAsync();


    //            [0]: "code"
    //[1]: "grant_type"
    //[2]: "redirect_uri"
    //[3]: "client_id"
    //[4]: "client_secret"

            return Ok(new Dictionary<string,object>{ ["access_token"] ="mytesttoken", ["token_type"]="bearer", ["expires_in"]= "300",["refresh_token"] ="refreshtoken", ["scope"]="read", ["uid"] ="testuid", ["info"] = new { name="poul", email="[email protected]"} });
           
        }
        [HttpPost("providers/IO-Board.DataIngestor/swagger/refresh")]
        public async Task<IActionResult> refresh()
        {
            var form = await Request.ReadFormAsync();

 

            return Ok(new Dictionary<string, object> { ["access_token"] = "mytesttoken", ["token_type"] = "bearer", ["expires_in"] = "300", ["refresh_token"] = "refreshtoken", ["scope"] = "read", ["uid"] = "testuid", ["info"] = new { name = "poul", email = "[email protected]" } });
            
        }

        [HttpPost("subscriptions/{subscriptionId}/providers/IO-Board.DataIngestor/collections/{collectionid}/hooks")]
        public async Task<IActionResult> CreateWebHook(string subscriptionId, string collectionid)
        {
            return Created($"https://io-board.eu.ngrok.io/subscriptions/{subscriptionId}/providers/IO-Board.DataIngestor/collections/{collectionid}/hooks/{Guid.NewGuid()}",new { hello="world"});
        }
        [HttpDelete("subscriptions/{subscriptionId}/providers/IO-Board.DataIngestor/collections/{collectionid}/hooks/{hookid}")]
        public async Task<IActionResult> DeleteWebHook(string subscriptionId, string collectionid)
        {
            return Ok();
        }


        [HttpGet("providers/IO-Board.DataIngestor/swagger")]
        public async Task<IActionResult> OpenAPI()
        {
            return Ok(
                new
                {
                    swagger = "2.0",
                    info =new
                    {
                        version="1.0",
                        title="Custom Flow",
                        description="My Custom Flow"
                    },
                    host ="io-board.eu.ngrok.io",
                    basePath = "/",
                    schemes = new[] { "https"},
                    securityDefinitions =new
                    {
                        oAuth = new
                        {
                            type="oauth2",
                            flow="accessCode",
                            authorizationUrl= "https://io-board.eu.ngrok.io/providers/IO-Board.DataIngestor/swagger/authorize",
                            tokenUrl = "https://io-board.eu.ngrok.io/providers/IO-Board.DataIngestor/swagger/token",
                            refreshUrl = "https://io-board.eu.ngrok.io/providers/IO-Board.DataIngestor/swagger/refresh",
                            scopes =new
                            {

                            }
                        }
                    },
                    consumes=new[] {"application/json"},
                    produces=new[] { "application/json"},
                    definitions =new 
                    {
                        WebhookPushResponse=new
                        {
                            type="object",
                            properties=new { }
                        },
                        WebhookRequestBody = new
                        {
                            type = "object",
                            properties = new { }
                        },
                        WebhookCreatingResponse = new
                        {
                            type = "object",
                            properties = new { }
                        }
                    },
                    paths = new Dictionary<string, object>
                    {
                        ["/subscriptions/{subscriptionId}/providers/IO-Board.DataIngestor/collections/{collectionid}/hooks"] = new Dictionary<string,object>
                        {
                            ["x-ms-notification-content"] = new
                            {
                                description = "Details for webhook",
                                schema=new Dictionary<string,string>
                                {
                                   ["$ref"] = "#/definitions/WebhookPushResponse" 
                                }
                            },
                            ["post"] = new Dictionary<string, object>
                            {
                                ["description"] = "Creates a IO-Board webhook",
                                ["summary"] = "Triggers when event is published",
                                ["operationId"] ="webhook-trigger",
                                ["x-ms-trigger"] = "single",
                                ["parameters"]= new[]
                                {
                                    new {
                                        name = "subscriptionId", 
                                        @in = "path",
                                        description="Subscription Id of the data hub",
                                        required=true,
                                        type="string"
                                    },
                                     new {
                                        name = "collectionid",
                                        @in = "path",
                                        description="collection Id of the data",
                                        required=true,
                                        type="string"
                                    },
                                       new {
                                        name = "Request body of webhook",
                                        @in = "body",
                                        description="this is the request body of the Webhook",
                                        schema = new Dictionary<string, string>
                                        {
                                            ["$ref"]="#/definitions/WebhookRequestBody"
                                        }
                                    } as object

                                },
                                ["responses"]=new Dictionary<string, object>
                                {
                                    ["201"]=new {
                                        description = "Created",
                                        schema = new Dictionary<string, string>
                                        {
                                            ["$ref"] = "#/definitions/WebhookCreationResponse"
                                        }
                                    }
                                }
                            }
                        }
                    }

                }
                );
        }


        [HttpPost("subscriptions/{subscriptionId}/providers/IO-Board.DataIngestor/collections/{collectionid}/events")]
        public async Task<IActionResult> PublishEvent(
            [FromServices] QueueProvider queueProvider,
            [FromServices] AuthorizationService authorizationService,
            [FromServices] CollectionContext collectionContext ,  
            [FromHeader]string authorization,
            Guid subscriptionId,
            string collectionid           
            )
        {

            var collectionMetadata = await memoryCache.GetOrCreateAsync(collectionid,(entry)=>
            {
                return collectionContext.Collections.Where(c => c.SubscriptionId == subscriptionId && c.Id == collectionid).FirstOrDefaultAsync();
            });

            var consumerInfo = await memoryCache.GetOrCreateAsync(collectionMetadata.ConsumerId, (entry) =>
            {
                return collectionContext.Consumers.Where(c => c.Id == collectionMetadata.ConsumerId ).FirstOrDefaultAsync();
            });

            var auth = await authorizationService.GetAuthorizationAsync(consumerInfo.Auth, authorization);

          

            using (var stream = new StreamReader(Request.Body))
            {
                var queue = await queueProvider.GetQueueAsync(subscriptionId, collectionid);

                var content = await stream.ReadToEndAsync();

                var eventd = new EventMessage
                {
                    Claims = auth.Claims.GroupBy(k => k.Type).ToDictionary(c => c.Key, v => v.Skip(1).Any() ? v.Select(v => v.Value).ToArray() : v.FirstOrDefault().Value as object),
                    Metadata = collectionMetadata.Metadata,
                    CollectionId = collectionid,
                    Content = content,
                    ContentMD5 = CreateMD5(content), 
                    RecordedAt = JsonPath.Select<DateTimeOffset?>(collectionMetadata.RecordedAtPath, content) ?? DateTimeOffset.UtcNow,
                    SubscriptionId = subscriptionId 
                };

                await queue.AddMessageAsync(eventd);

                return Accepted();
            }
        }
        public static string CreateMD5(string input)
        {
            // Use input string to calculate MD5 hash
            using (System.Security.Cryptography.MD5 md5 = System.Security.Cryptography.MD5.Create())
            {
                byte[] inputBytes = System.Text.Encoding.ASCII.GetBytes(input);
                byte[] hashBytes = md5.ComputeHash(inputBytes);

                // Convert the byte array to hexadecimal string
                StringBuilder sb = new StringBuilder();
                for (int i = 0; i < hashBytes.Length; i++)
                {
                    sb.Append(hashBytes[i].ToString("X2"));
                }
                return sb.ToString();
            }
        }
    }


}