//
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.
//

import org.apache.asterix.bad.lang.statement.BrokerDropStatement;
import org.apache.asterix.bad.lang.statement.ChannelDropStatement;
import org.apache.asterix.bad.lang.statement.ChannelSubscribeStatement;
import org.apache.asterix.bad.lang.statement.ChannelUnsubscribeStatement;
import org.apache.asterix.bad.lang.statement.CreateBrokerStatement;
import org.apache.asterix.bad.lang.statement.AbstractCreateChannelStatement;
import org.apache.asterix.bad.lang.statement.CreateRepetitiveChannelStatement;
import org.apache.asterix.bad.lang.statement.CreateContinuousChannelStatement;
import org.apache.asterix.bad.lang.statement.CreateProcedureStatement;
import org.apache.asterix.bad.lang.statement.ExecuteProcedureStatement;
import org.apache.asterix.bad.lang.statement.ProcedureDropStatement;
import org.apache.asterix.lang.sqlpp.parser.ParseException;
import org.apache.asterix.lang.sqlpp.parser.SqlppParseException;
import org.apache.asterix.lang.sqlpp.parser.Token;

@new
<DEFAULT,IN_DBL_BRACE>
TOKEN [IGNORE_CASE]:
{
  <BROKER : "broker">
  |  <CHANNEL : "channel">
  |  <PROCEDURE : "procedure">
  |  <SUBSCRIBE : "subscribe">
  |  <ACTIVE : "active">
  | <UNSUBSCRIBE : "unsubscribe">
  | <REPETITIVE : "repetitive">
  | <CONTINUOUS : "continuous">
  | <PERIOD : "period">
  | <PUSH : "push">
}

@merge
Statement SingleStatement() throws ParseException:
{
  // merge area 1
  before:
  after:
}
{
  (
    // merge area 2
    before:
    after:    | stmt = ChannelSubscriptionStatement() | stmt = ProcedureExecution())
  {
    // merge area 3
  }
}

@merge
Statement CreateStatement() throws ParseException:
{
  // merge area 1
  before:
  after:
}
{
  (
    // merge area 2
    before:
    after:    | stmt = CreateChannelStatement(startToken)
              | stmt = CreateBrokerStatement(startToken)
              | stmt = CreateProcedureStatement()
              | stmt = ActiveStatementSpecification(startToken)
  )
  {
    // merge area 3
  }
}

@merge
Statement DropStatement() throws ParseException:
{
  // merge area 1
  before:
  after:
}
{
  (
    // merge area 2
    before:
    after:    | stmt = DropChannelStatement(startToken)
              | stmt = DropBrokerStatement(startToken)
              | stmt = DropProcedureStatement(startToken)
  )
  {
    // merge area 3
  }
}

@new
Statement ActiveStatementSpecification(Token startStmtToken) throws ParseException:
{
    Statement stmt = null;
}
{
  "ACTIVE"
   stmt = ActiveDatasetSpecification(startStmtToken)
  {
      return stmt;
  }
}

@new
DatasetDecl ActiveDatasetSpecification(Token startStmtToken) throws ParseException:
{
  Pair<DataverseName,Identifier> nameComponents = null;
  TypeExpression activeRecordTypeExpr = new TypeReferenceExpression(
                new Pair<DataverseName,Identifier>(DataverseName.createBuiltinDataverseName("Metadata"), new Identifier("ActiveRecordType")));
  TypeExpression datasetTypeExpr = null;
  boolean ifNotExists = false;
  Map<String,String> properties = null;
  Pair<List<Integer>, List<List<String>>> primaryKeyFields = null;
  Map<String,String> hints = new HashMap<String,String>();
  DatasetDecl stmt = null;
  boolean autogenerated = false;
  RecordConstructor withRecord = null;

}
{
    Dataset() nameComponents = QualifiedName()
    datasetTypeExpr = DatasetTypeSpecification()
    ifNotExists = IfNotExists()
    primaryKeyFields = PrimaryKey()
    (<AUTOGENERATED> { autogenerated = true; } )?
    ( <HINTS> hints = Properties() )?
    ( <WITH> withRecord = RecordConstructor() )?
      {
        // TODO: add filters on meta records
        InternalDetailsDecl idd = new InternalDetailsDecl(primaryKeyFields.second,
                                                          primaryKeyFields.first,
                                                          autogenerated,
                                                          null);
        try{
        stmt = new DatasetDecl(nameComponents.first,
                                   nameComponents.second,
                                   datasetTypeExpr,
                                   activeRecordTypeExpr,
                                   hints,
                                   DatasetType.INTERNAL,
                                   idd,
                                   withRecord,
                                   ifNotExists);

        } catch (CompilationException e){
           throw new SqlppParseException(getSourceLocation(startStmtToken), e.getMessage());
        }
      }
    {
      return addSourceLocation(stmt, startStmtToken);
    }
}


@new
AbstractCreateChannelStatement CreateChannelStatement(Token startStmtToken) throws ParseException:
{
  AbstractCreateChannelStatement ccs = null;
}
{
  (
    <REPETITIVE> { ccs = CreateRepetitiveChannel(startStmtToken); }
    | <CONTINUOUS> { ccs = CreateContinuousChannel(startStmtToken); }
  )
  {
    return ccs;
  }
}

@new
CreateRepetitiveChannelStatement CreateRepetitiveChannel(Token startStmtToken) throws ParseException:
{
  Pair<DataverseName,Identifier> nameComponents = null;
  FunctionSignature appliedFunction = null;
  CreateRepetitiveChannelStatement ccs = null;
  String fqFunctionName = null;
  Expression period = null;
  boolean push = false;
}
{
  ( <PUSH> { push = true; } )?
  <CHANNEL>  nameComponents = QualifiedName()
  <USING> appliedFunction = FunctionSignature()
  <PERIOD> period = FunctionCallExpr()
  {
    ccs = new CreateRepetitiveChannelStatement(nameComponents.first,
                                 nameComponents.second, appliedFunction, period, push);
  }
  {
    return ccs;
  }
}


@new
CreateContinuousChannelStatement CreateContinuousChannel(Token startStmtToken) throws ParseException:
{
  // Channel related
  CreateContinuousChannelStatement ccs = null;
  Expression period = null;
  boolean push = false;
  boolean ifNotExists = false;
  // Function related
  FunctionSignature signature;
  String functionBody;
  Expression functionBodyExpr;
  Token beginPos;
  Token endPos;
  FunctionName fctName = null;
  TypeExpression returnType = null;
  List<Pair<VarIdentifier,TypeExpression>> params = null;
  DataverseName currentDataverse = defaultDataverse;
}
{
     ( <PUSH> { push = true; } )?
     <CHANNEL>
     fctName = FunctionName()
     {
        defaultDataverse = fctName.dataverse;
     }
     ifNotExists = IfNotExists()
     params = FunctionParameters()
     <PERIOD> period = FunctionCallExpr()
    <LEFTBRACE>
      {
        createNewScope();
        beginPos = token;
      }
      functionBodyExpr = FunctionBody()
      returnType = FunctionReturnType()
    <RIGHTBRACE>
    {
      endPos = token;
      functionBody = extractFragment(beginPos.beginLine, beginPos.beginColumn, endPos.beginLine, endPos.beginColumn);
      signature = new FunctionSignature(fctName.dataverse, fctName.function, params.size());
      getCurrentScope().addFunctionDescriptor(signature, false);
      removeCurrentScope();
      defaultDataverse = currentDataverse;
      ensureNoTypeDeclsInFunction(fctName.function, params, returnType, startStmtToken);
      CreateFunctionStatement stmt = new CreateFunctionStatement(signature, params, functionBody, functionBodyExpr, false);
      ccs =  new CreateContinuousChannelStatement(fctName.dataverse, new Identifier(fctName.function), period, push, stmt);
      return addSourceLocation(ccs, startStmtToken);
    }
}

@new
CreateProcedureStatement CreateProcedureStatement() throws ParseException:
{
  FunctionName fctName = null;
  FunctionSignature signature;
  List<Pair<VarIdentifier,TypeExpression>> paramList = new ArrayList<Pair<VarIdentifier,TypeExpression>>();
  List<Integer> paramIds = new ArrayList<Integer>();
  String functionBody;
  Token beginPos;
  Token endPos;
  Statement functionBodyExpr;
  Expression period = null;
  DataverseName currentDataverse = defaultDataverse;
  createNewScope();
}
{
     <PROCEDURE> fctName = FunctionName()
     {
        defaultDataverse = fctName.dataverse;
     }
     paramList = FunctionParameters()
    <LEFTBRACE>
  {
    for (Pair<VarIdentifier,TypeExpression> param : paramList)
    {
      VarIdentifier v = new VarIdentifier(param.getFirst().toString());
      getCurrentScope().addNewVarSymbolToScope(v);
      paramIds.add(v.getId());
    }
    beginPos = token;
  }
  functionBodyExpr = SingleStatement() <RIGHTBRACE>
    {
      endPos = token;
      functionBody = extractFragment(beginPos.beginLine, beginPos.beginColumn, endPos.beginLine, endPos.beginColumn);
      signature = new FunctionSignature(fctName.dataverse, fctName.function, paramList.size());
      removeCurrentScope();
      defaultDataverse = currentDataverse;
    }
  (<PERIOD> period = FunctionCallExpr())?
  {
  List<VarIdentifier> paramListVariablesOnly = new ArrayList<VarIdentifier>();
  for(Pair<VarIdentifier,TypeExpression> p: paramList){
    paramListVariablesOnly.add(p.first);
  }
  return new CreateProcedureStatement(signature, paramListVariablesOnly, paramIds, functionBody, period);
  }
}

@new
ExecuteProcedureStatement ProcedureExecution() throws ParseException:
{
  ExecuteProcedureStatement callExpr;
  List<Expression> argList = new ArrayList<Expression>();
  Expression tmp;
  int arity = 0;
  FunctionName funcName = null;
  String hint = null;
}
{
  "execute"
  funcName = FunctionName()
  <LEFTPAREN> (tmp = Expression()
    {
      argList.add(tmp);
      arity ++;
    }
  (<COMMA> tmp = Expression()
    {
      argList.add(tmp);
      arity++;
    }
  )*)? <RIGHTPAREN>
    {
      String fqFunctionName =  funcName.function;
      return new ExecuteProcedureStatement(funcName.dataverse, fqFunctionName, arity, argList);
    }
}

@new
CreateBrokerStatement CreateBrokerStatement(Token startStmtToken) throws ParseException:
{
  CreateBrokerStatement cbs = null;
  Pair<DataverseName,Identifier> name = null;
  String endPoint = null;
  RecordConstructor withRecord = null;
}
{
  (
    <BROKER>  name = QualifiedName()
    <AT>  endPoint = StringLiteral()
    ( <WITH> withRecord = RecordConstructor() )?
    {
      try {
        cbs = new CreateBrokerStatement(name.first, name.second, endPoint, withRecord);
      } catch (CompilationException e) {
        throw new SqlppParseException(getSourceLocation(startStmtToken), e.getMessage());
      }
    }
  )
    {
      return cbs;
    }
}

@new
Statement ChannelSubscriptionStatement() throws ParseException:
{
  Statement stmt = null;
  Pair<DataverseName,Identifier> nameComponents = null;
  List<Expression> argList = new ArrayList<Expression>();
  Expression tmp = null;
  String id = null;
  String subscriptionId = null;
  Pair<DataverseName,Identifier> brokerName = null;
}
{
  (
  <SUBSCRIBE> <TO> nameComponents = QualifiedName()
   <LEFTPAREN> (tmp = Expression()
   {
      argList.add(tmp);
   }
   (<COMMA> tmp = Expression()
   {
      argList.add(tmp);
   }
   )*)? <RIGHTPAREN> <ON> brokerName = QualifiedName()
   {
      stmt = new ChannelSubscribeStatement(nameComponents.first, nameComponents.second, argList, getVarCounter(), brokerName.first, brokerName.second, subscriptionId);
   }
   | <UNSUBSCRIBE> id = StringLiteral() <FROM> nameComponents = QualifiedName()
      {
        VariableExpr varExp = new VariableExpr(new VarIdentifier("$subscriptionPlaceholder"));
        getCurrentScope().addNewVarSymbolToScope(varExp.getVar());
        stmt = new ChannelUnsubscribeStatement(varExp, nameComponents.first, nameComponents.second, id, getVarCounter());
      }
     | "change" "subscription" subscriptionId = StringLiteral()  <ON> nameComponents = QualifiedName()
       <LEFTPAREN> (tmp = Expression()
       {
         argList.add(tmp);
       }
       (<COMMA> tmp = Expression()
       {
         argList.add(tmp);
       }
       )*)? <RIGHTPAREN>
        <TO> brokerName = QualifiedName()
      {
        stmt = new ChannelSubscribeStatement(nameComponents.first, nameComponents.second, argList, getVarCounter(), brokerName.first, brokerName.second, subscriptionId);
      }
    )
    {
      return stmt;
    }
}

@new
ChannelDropStatement DropChannelStatement(Token startStmtToken) throws ParseException:
{
  ChannelDropStatement stmt = null;
  Pair<DataverseName,Identifier> pairId = null;
  boolean ifExists = false;
}
{
  <CHANNEL> pairId = QualifiedName() ifExists = IfExists()
  {
    stmt = new ChannelDropStatement(pairId.first, pairId.second, ifExists);
    return addSourceLocation(stmt, startStmtToken);
  }
}

@new
BrokerDropStatement DropBrokerStatement(Token startStmtToken) throws ParseException:
{
  BrokerDropStatement stmt = null;
  Pair<DataverseName,Identifier> pairId = null;
  boolean ifExists = false;
}
{
  <BROKER> pairId = QualifiedName() ifExists = IfExists()
  {
    stmt = new BrokerDropStatement(pairId.first, pairId.second, ifExists);
    return addSourceLocation(stmt, startStmtToken);
  }
}

@new
ProcedureDropStatement DropProcedureStatement(Token startStmtToken) throws ParseException:
{
  ProcedureDropStatement stmt = null;
  FunctionSignature funcSig = null;
  boolean ifExists = false;
}
{
  <PROCEDURE> funcSig = FunctionSignature() ifExists = IfExists()
  {
    stmt = new ProcedureDropStatement(funcSig, ifExists);
    return addSourceLocation(stmt, startStmtToken);
  }
}