content: -- | A utility for transforming Avro-compliant JSON into other models
  
  module Hydra.Ext.Tools.AvroWorkflows (
    TransformWorkflow(..),
    JsonPayloadFormat(..),
    TermEncoder(..),
    LastMile(..),
    defaultTinkerpopAnnotations,
    exampleGraphsonContext,
    examplePgSchema,
    executeAvroTransformWorkflow,
    propertyGraphGraphsonLastMile,
    propertyGraphJsonLastMile,
    rdfDescriptionsToNtriples,
    shaclRdfLastMile,
    typedTermToShaclRdf,
    transformAvroJsonDirectory,
  ) where
  
  import Hydra.Kernel
  import Hydra.Tools.Monads
  import Hydra.Dsl.Annotations
  import qualified Hydra.Ext.Org.Apache.Avro.Schema as Avro
  import qualified Hydra.Json as Json
  import Hydra.Ext.Org.Json.Coder
  import Hydra.Extract.Json
  import Hydra.Staging.Json.Serde
  import Hydra.Ext.Staging.Avro.Coder
  import Hydra.Ext.Staging.Avro.SchemaJson
  import Hydra.Ext.Staging.Pg.Graphson.Utils
  import Hydra.Ext.Staging.Pg.Coder
  import qualified Hydra.Ext.Staging.Shacl.Coder as Shacl
  import qualified Hydra.Ext.Org.W3.Rdf.Syntax as Rdf
  import qualified Hydra.Ext.Staging.Rdf.Utils as RdfUt
  import qualified Hydra.Pg.Model as PG
  import qualified Hydra.Pg.Mapping as PGM
  import Hydra.Ext.Staging.Rdf.Serde
  import Hydra.Sources.Kernel.Types.Core
  import Hydra.Ext.Staging.Pg.Graphson.Coder
  import Hydra.Pg.Graphson.Syntax as G
  import Hydra.Ext.Staging.Pg.Utils
  
  import qualified Control.Monad as CM
  import qualified Data.Either as E
  import qualified Data.List as L
  import qualified Data.Map as M
  import qualified Data.Set as S
  import qualified Data.Maybe as Y
  import System.IO
  import System.FilePath
  import System.FilePath.Posix
  import System.Directory
  
  
  data JsonPayloadFormat = Json | Jsonl
  
  type TermEncoder x = Term -> Graph -> Flow Graph [x]
  
  -- | A convenience for transformAvroJsonDirectory, bundling all of the input parameters together as a workflow
  executeAvroTransformWorkflow :: LastMile Graph x -> TransformWorkflow -> IO ()
  executeAvroTransformWorkflow lastMile (TransformWorkflow name schemaSpec srcDir destDir) = do
      schemaPath <- case schemaSpec of
        SchemaSpecFile p -> pure p
        _ -> fail "unsupported schema spec"
      putStrLn $ "Executing workflow " ++ show name ++ ":"
      transformAvroJsonDirectory lastMile schemaPath srcDir destDir
  
  -- Replace all lists with sets, for better query performance.
  -- This is a last-mile step which breaks type/term conformance
  -- (a more robust solution would modify the target language in the SHACL coder, so that list types are also transformed to set types).
  listsToSets :: Term -> Term
  listsToSets = rewriteTerm mapExpr
    where
      mapExpr recurse = recurse . replaceLists
      replaceLists term = case term of
        TermList els -> TermSet $ S.fromList els
        _ -> term
  
  rdfDescriptionsToNtriples :: [Rdf.Description] -> String
  rdfDescriptionsToNtriples = rdfGraphToNtriples . RdfUt.descriptionsToGraph
  
  shaclRdfLastMile :: LastMile Graph Rdf.Description
  shaclRdfLastMile = LastMile typedTermToShaclRdf (pure . rdfDescriptionsToNtriples) "nt"
  
  typedTermToShaclRdf :: Type -> Flow Graph (Term -> Graph -> Flow Graph [Rdf.Description])
  typedTermToShaclRdf _ = pure encode
    where
      encode term graph = do
          elDescs <- CM.mapM encodeElement $ M.elems $ graphElements graph
          termDescs <- encodeBlankTerm
          return $ L.concat (termDescs:elDescs)
        where
          encodeElement el = do
            let subject = Rdf.ResourceIri $ RdfUt.nameToIri $ bindingName el
            Shacl.encodeTerm subject $ listsToSets $ bindingTerm el
          encodeBlankTerm = if notInGraph
            then do
              subject <- RdfUt.nextBlankNode
              Shacl.encodeTerm subject $ listsToSets term
            else pure []
          notInGraph = L.null $ L.filter (\e -> bindingTerm e == term) $ M.elems $ graphElements graph
  
  transformAvroJson :: JsonPayloadFormat -> AvroHydraAdapter -> LastMile Graph x -> FilePath -> FilePath -> IO ()
  transformAvroJson format adapter lastMile inFile outFile = do
      putStr $ "\t" ++ inFile ++ " --> "
      contents <- readFile inFile
      let entities = case format of
            Json -> [contents]
            Jsonl -> L.filter (not . L.null) $ lines contents
      lmEncoder <- flowToIo hydraCoreGraph $ lastMileEncoder lastMile (adapterTarget adapter)
      descs <- L.concat <$> flowToIo hydraCoreGraph (CM.zipWithM (jsonToTarget inFile adapter lmEncoder) [1..] entities)
      result <- flowToIo hydraCoreGraph $ lastMileSerializer lastMile descs
      writeFile outFile result
      putStrLn $ outFile ++ " (" ++ descEntities entities ++ ")"
    where
      descEntities entities = if L.length entities == 1 then "1 entity" else show (L.length entities) ++ " entities"
  
      jsonToTarget inFile adapter lmEncoder index payload = case stringToJsonValue payload of
          Left msg -> fail $ "Failed to read JSON payload #" ++ show index ++ " in file " ++ inFile ++ ": " ++ msg
          Right json -> withState emptyAvroEnvironment $ do
            -- TODO; the core graph is neither the data nor the schema graph
            let dataGraph = hydraCoreGraph
            let schemaGraph = Just hydraCoreGraph
  
            term <- coderEncode (adapterCoder adapter) json
            env <- getState
            let graph = elementsToGraph dataGraph schemaGraph (M.elems $ avroEnvironmentElements env)
            withState hydraCoreGraph $ lmEncoder term graph
  
  -- | Given a payload format (one JSON object per file, or one per line),
  --   a path to an Avro *.avsc schema, a path to a source directory containing JSON files conforming to the schema,
  --   and a path to a destination directory, map each input file to a corresponding output file in the
  --   destination directory. This transformation is sensitive to Hydra-specific annotations (primaryKey/foreignKey)
  --   in the Avro schema, which tell Hydra which objects to treat as elements and which fields are references to elements.
  transformAvroJsonDirectory :: LastMile Graph x -> FilePath -> FilePath -> FilePath -> IO ()
  transformAvroJsonDirectory lastMile schemaPath srcDir destDir = do
      createDirectoryIfMissing True destDir
      schemaStr <- readFile schemaPath
      adapter <- flowToIo () $ loadAdapter schemaStr
      paths <- getDirectoryContents srcDir
      conf <- CM.mapM (transformFile adapter) paths
      return ()
    where
      loadAdapter schemaStr = do
        avroSchema <- coderDecode avroSchemaStringCoder schemaStr
        withState emptyAvroEnvironment $ avroHydraAdapter avroSchema
  
      transformFile adapter srcFile = do
        case jsonPayloadFormat srcFile of
          Nothing -> return False
          Just format -> do
            let destFile = replaceExtension srcFile (lastMileFileExtension lastMile)
            transformAvroJson format adapter lastMile (combine srcDir srcFile) (combine destDir destFile)
            return True
  
      jsonPayloadFormat fileName = if ext == ".json"
          then Just Json
          else if ext == ".jsonl"
          then Just Jsonl
          else Nothing
        where
          ext = takeExtension fileName
  