<commit_before>import { HttpClient } from '@angular/common/http'
import { Injectable } from '@angular/core'
import * as AvroSchema from 'avsc'
import * as KafkaRest from 'kafka-rest'
import YAML from 'yaml'

import { DefaultSchemaSpecEndpoint } from '../../../../assets/data/defaultConfig'
import { ConfigKeys } from '../../../shared/enums/config'
import { AnswerValueExport } from '../../../shared/models/answer'
import { QuestionnaireMetadata } from '../../../shared/models/assessment'
import { CompletionLogValueExport } from '../../../shared/models/completion-log'
import { EventValueExport } from '../../../shared/models/event'
import {
  KeyExport,
  SchemaMetadata,
  SchemaType
} from '../../../shared/models/kafka'
import { Task } from '../../../shared/models/task'
import { ApplicationTimeZoneValueExport } from '../../../shared/models/timezone'
import { getTaskType } from '../../../shared/utilities/task-type'
import { getSeconds } from '../../../shared/utilities/time'
import { QuestionnaireService } from '../config/questionnaire.service'
import { RemoteConfigService } from '../config/remote-config.service'
import { SubjectConfigService } from '../config/subject-config.service'
import { LogService } from '../misc/log.service'

@Injectable()
export class SchemaService {
  URI_schema: string = '/schema/subjects/'
  URI_version: string = '/versions/'
  private schemas: {
    [key: string]: [Promise<SchemaMetadata>, Promise<SchemaMetadata>]
  } = {}

  constructor(
    public questionnaire: QuestionnaireService,
    private config: SubjectConfigService,
    private http: HttpClient,
    private logger: LogService,
    private remoteConfig: RemoteConfigService
  ) {}

  getMetaData(type, task?: Task): Promise<QuestionnaireMetadata> {
    switch (type) {
      case SchemaType.ASSESSMENT:
        return this.questionnaire
          .getAssessment(getTaskType(task), task)
          .then(assessment => assessment.questionnaire)
      default:
        return Promise.resolve({ name: type, avsc: 'questionnaire' })
    }
  }

  getKafkaObjectKey() {
    return Promise.all([
      this.config.getSourceID(),
      this.config.getProjectName(),
      this.config.getParticipantLogin()
    ])
      .then(([sourceId, projectName, participantName]) => ({
        sourceId,
        userId: participantName.toString(),
        projectId: projectName
      }))
      .then(observationKey => observationKey as KeyExport)
  }

  getKafkaObjectValue(type, payload) {
    switch (type) {
      case SchemaType.ASSESSMENT:
        const Answer: AnswerValueExport = {
          name: payload.task.name,
          version: payload.data.configVersion,
          answers: payload.data.answers,
          time: payload.data.time,
          timeCompleted: payload.data.timeCompleted,
          timeNotification: getSeconds({ milliseconds: payload.task.timestamp })
        }
        return Answer
      case SchemaType.COMPLETION_LOG:
        const CompletionLog: CompletionLogValueExport = {
          name: payload.name,
          time: getSeconds({ milliseconds: this.getUniqueTimeNow() }),
          timeNotification: getSeconds({
            milliseconds: payload.timeNotification
          }),
          completionPercentage: { double: payload.percentage }
        }
        return CompletionLog
      case SchemaType.TIMEZONE:
        const ApplicationTimeZone: ApplicationTimeZoneValueExport = {
          time: getSeconds({ milliseconds: this.getUniqueTimeNow() }),
          offset: getSeconds({ minutes: new Date().getTimezoneOffset() })
        }
        return ApplicationTimeZone
      case SchemaType.APP_EVENT:
        const Event: EventValueExport = {
          time: getSeconds({ milliseconds: this.getUniqueTimeNow() }),
          eventType: payload.eventType
        }
        return Event
    }
  }

  getAvroObject(schema, value) {
    const options = { wrapUnions: true }
    return AvroSchema.parse(schema, options).clone(value, options)
  }

  convertToAvro(kafkaObject, topic, baseURI) {
    if (!this.schemas[topic]) {
      const schemaKeyAndValue: [
        Promise<SchemaMetadata>,
        Promise<SchemaMetadata>
      ] = [
        this.getLatestKafkaSchemaVersion(topic + '-key', 'latest', baseURI),
        this.getLatestKafkaSchemaVersion(topic + '-value', 'latest', baseURI)
      ]
      this.schemas[topic] = schemaKeyAndValue
    }
    return Promise.all(this.schemas[topic]).then(
      ([keySchemaMetadata, valueSchemaMetadata]) => {
        if (keySchemaMetadata && valueSchemaMetadata) {
          const key = JSON.parse(keySchemaMetadata.schema)
          const value = JSON.parse(valueSchemaMetadata.schema)
          const schemaId = new KafkaRest.AvroSchema(key)
          const schemaInfo = new KafkaRest.AvroSchema(value)
          const payload = {
            key: this.getAvroObject(key, kafkaObject.key),
            value: this.getAvroObject(value, kafkaObject.value)
          }
          return { schemaId, schemaInfo, payload }
        } else {
          Promise.reject()
        }
      }
    )
  }

  getKafkaTopic(name, avsc) {
    const type = name.toLowerCase()
    const defaultTopic = `${avsc}_${name}`
    return this.remoteConfig
      .read()
      .then(config =>
        config.getOrDefault(
          ConfigKeys.KAFKA_SPECIFICATION_URL,
          DefaultSchemaSpecEndpoint
        )
      )
      .then(url => this.http.get(url).toPromise())
      .then(res => {
        const schemaSpecs = YAML.parse(atob(res['content'])).data
        const topic = schemaSpecs.find(t => t.type.toLowerCase() == type).topic
        if (topic) return topic
        else return Promise.reject()
      })
      .catch(e => defaultTopic)
  }

  getLatestKafkaSchemaVersion(
    questionName,
    version,
    endPoint
  ): Promise<SchemaMetadata> {
    const versionStr = this.URI_version + version
    const uri = endPoint + this.URI_schema + questionName + versionStr

    return this.http
      .get(uri)
      .toPromise()
      .catch(e =>
        // TODO: add fallback for error
        this.logger.error('Failed to get latest Kafka schema versions', e)
      )<commit_msg>fix: throw readable error if schema is not found<commit_after>import { HttpClient } from '@angular/common/http'
import { Injectable } from '@angular/core'
import * as AvroSchema from 'avsc'
import * as KafkaRest from 'kafka-rest'
import YAML from 'yaml'

import { DefaultSchemaSpecEndpoint } from '../../../../assets/data/defaultConfig'
import { ConfigKeys } from '../../../shared/enums/config'
import { AnswerValueExport } from '../../../shared/models/answer'
import { QuestionnaireMetadata } from '../../../shared/models/assessment'
import { CompletionLogValueExport } from '../../../shared/models/completion-log'
import { EventValueExport } from '../../../shared/models/event'
import {
  KeyExport,
  SchemaMetadata,
  SchemaType
} from '../../../shared/models/kafka'
import { Task } from '../../../shared/models/task'
import { ApplicationTimeZoneValueExport } from '../../../shared/models/timezone'
import { getTaskType } from '../../../shared/utilities/task-type'
import { getSeconds } from '../../../shared/utilities/time'
import { QuestionnaireService } from '../config/questionnaire.service'
import { RemoteConfigService } from '../config/remote-config.service'
import { SubjectConfigService } from '../config/subject-config.service'
import { LogService } from '../misc/log.service'

@Injectable()
export class SchemaService {
  URI_schema: string = '/schema/subjects/'
  URI_version: string = '/versions/'
  private schemas: {
    [key: string]: [Promise<SchemaMetadata>, Promise<SchemaMetadata>]
  } = {}

  constructor(
    public questionnaire: QuestionnaireService,
    private config: SubjectConfigService,
    private http: HttpClient,
    private logger: LogService,
    private remoteConfig: RemoteConfigService
  ) {}

  getMetaData(type, task?: Task): Promise<QuestionnaireMetadata> {
    switch (type) {
      case SchemaType.ASSESSMENT:
        return this.questionnaire
          .getAssessment(getTaskType(task), task)
          .then(assessment => assessment.questionnaire)
      default:
        return Promise.resolve({ name: type, avsc: 'questionnaire' })
    }
  }

  getKafkaObjectKey() {
    return Promise.all([
      this.config.getSourceID(),
      this.config.getProjectName(),
      this.config.getParticipantLogin()
    ])
      .then(([sourceId, projectName, participantName]) => ({
        sourceId,
        userId: participantName.toString(),
        projectId: projectName
      }))
      .then(observationKey => observationKey as KeyExport)
  }

  getKafkaObjectValue(type, payload) {
    switch (type) {
      case SchemaType.ASSESSMENT:
        const Answer: AnswerValueExport = {
          name: payload.task.name,
          version: payload.data.configVersion,
          answers: payload.data.answers,
          time: payload.data.time,
          timeCompleted: payload.data.timeCompleted,
          timeNotification: getSeconds({ milliseconds: payload.task.timestamp })
        }
        return Answer
      case SchemaType.COMPLETION_LOG:
        const CompletionLog: CompletionLogValueExport = {
          name: payload.name,
          time: getSeconds({ milliseconds: this.getUniqueTimeNow() }),
          timeNotification: getSeconds({
            milliseconds: payload.timeNotification
          }),
          completionPercentage: { double: payload.percentage }
        }
        return CompletionLog
      case SchemaType.TIMEZONE:
        const ApplicationTimeZone: ApplicationTimeZoneValueExport = {
          time: getSeconds({ milliseconds: this.getUniqueTimeNow() }),
          offset: getSeconds({ minutes: new Date().getTimezoneOffset() })
        }
        return ApplicationTimeZone
      case SchemaType.APP_EVENT:
        const Event: EventValueExport = {
          time: getSeconds({ milliseconds: this.getUniqueTimeNow() }),
          eventType: payload.eventType
        }
        return Event
    }
  }

  getAvroObject(schema, value): any {
    const options = { wrapUnions: true }
    return AvroSchema.parse(schema, options).clone(value, options)
  }

  convertToAvro(kafkaObject, topic, baseURI): Promise<any> {
    if (!this.schemas[topic]) {
      this.schemas[topic] = [
        this.getLatestKafkaSchemaVersion(topic + '-key', 'latest', baseURI),
        this.getLatestKafkaSchemaVersion(topic + '-value', 'latest', baseURI)
      ]
    }
    return Promise.all(this.schemas[topic])
      .then(
        ([keySchemaMetadata, valueSchemaMetadata]) => {
          const key = JSON.parse(keySchemaMetadata.schema)
          const value = JSON.parse(valueSchemaMetadata.schema)
          const schemaId = new KafkaRest.AvroSchema(key)
          const schemaInfo = new KafkaRest.AvroSchema(value)
          const payload = {
            key: this.getAvroObject(key, kafkaObject.key),
            value: this.getAvroObject(value, kafkaObject.value)
          }
          return { schemaId, schemaInfo, payload }
        }
      )
      .catch(e => {
        this.schemas[topic] = null
        throw e
      })
  }

  getKafkaTopic(name, avsc) {
    const type = name.toLowerCase()
    const defaultTopic = `${avsc}_${name}`
    return this.remoteConfig
      .read()
      .then(config =>
        config.getOrDefault(
          ConfigKeys.KAFKA_SPECIFICATION_URL,
          DefaultSchemaSpecEndpoint
        )
      )
      .then(url => this.http.get(url).toPromise())
      .then(res => {
        const schemaSpecs = YAML.parse(atob(res['content'])).data
        const topic = schemaSpecs.find(t => t.type.toLowerCase() == type).topic
        if (topic) return topic
        else return Promise.reject()
      })
      .catch(e => defaultTopic)
  }

  getLatestKafkaSchemaVersion(
    questionName,
    version,
    endPoint
  ): Promise<SchemaMetadata> {
    const versionStr = this.URI_version + version
    const uri = endPoint + this.URI_schema + questionName + versionStr

    return this.http
      .get(uri)
      .toPromise()
      .catch(e => {
        throw this.logger.error('Failed to get latest Kafka schema versions', e)
      })
{'fraction_non_alphanumeric': 0.12323016134343101, 'fraction_numerical': 0.0, 'mean_word_length': 2.710751374465486, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2485377', 'n_tokens_mistral': 3367, 'n_tokens_neox': 3347, 'n_words': 724}