<commit_before>/* DO NOT EDIT: File is auto-generated */
package <%= to_package(data_module.sync.qualified_sync_ejb_name) %>;

@javax.ejb.Singleton( name = iris.acal.server.service.master.SynchronizationService.NAME )
@javax.ejb.TransactionManagement( javax.ejb.TransactionManagementType.BEAN )
@javax.ejb.ConcurrencyManagement( javax.ejb.ConcurrencyManagementType.BEAN )
public class <%= data_module.sync.sync_ejb_name %>
  extends iris.syncrecord.server.service.StandardSyncService
  implements <%= data_module.service_by_name(:SynchronizationService).ejb.qualified_service_name %>
{
  private static final java.util.logging.Logger LOG = java.util.logging.Logger.getLogger( <%= data_module.sync.sync_ejb_name %>.class.getName() );

  private static final int SYNC_ERROR_CODE = 42;

  @javax.persistence.PersistenceContext( unitName = <%= data_module.repository.jpa.qualified_unit_descriptor_name %>.NAME )
  private javax.persistence.EntityManager _entityManager;
  @javax.annotation.Resource
  private javax.transaction.UserTransaction _transaction;
<% data_module.sync.entities_to_synchronize.each do |entity| -%>
  @javax.inject.Inject
  private <%= entity.dao.jpa.qualified_dao_service_name %> _<%= entity.data_module.name %><%= entity.name %>Repository;
<% end -%>
  @javax.annotation.Resource
  private javax.transaction.TransactionSynchronizationRegistry _registry;
  @javax.inject.Inject
  private org.realityforge.replicant.server.EntityMessageEndpoint _endpoint;
  @javax.inject.Inject
  private <%= data_module.service_by_name(:SynchronizationContext).ejb.qualified_service_name %> _context;

  private volatile boolean _inSync;

  @java.lang.Override
  @javax.annotation.Nonnull
  public iris.syncrecord.server.data_type.SyncStatusDTO synchronizeDataSource( @javax.annotation.Nonnull final String dataSourceCode )
  {
    if ( !_inSync )
    {
      _inSync = true;
      try
      {
        if ( isFeatureEnabled( dataSourceCode + ".syncEnabled" ) )
        {
          return performSynchronization( dataSourceCode );
        }
        else
        {
          return createDisabledSyncStatus( dataSourceCode );
        }
      }
      finally
      {
        _inSync = false;
      }
    }
    else
    {
      return createAlreadyActiveSyncStatus( dataSourceCode );
    }
  }

  private iris.syncrecord.server.data_type.SyncStatusDTO performSynchronization( @javax.annotation.Nonnull final String dataSourceCode )
  {
    final iris.syncrecord.server.service.SynchronizationRecorder recorder = startSynchronization( dataSourceCode );

<% data_module.sync.entities_to_synchronize.each do |entity| -%>
    recorder.incMetric( "<%= entity.data_module.name %>.<%= entity.name %>.AddCount", 0 );
    recorder.incMetric( "<%= entity.data_module.name %>.<%= entity.name %>.RemoveCount", 0 );
    recorder.incMetric( "<%= entity.data_module.name %>.<%= entity.name %>.UpdateCount", 0 );
<% end -%>

    log( dataSourceCode, "Synchronizing data" );
    try
    {
<% data_module.sync.entities_to_synchronize.reverse.each do |entity| -%>
      {
        log( dataSourceCode, "Synchronizing <%= entity.data_module.name %>.<%= entity.name %> removals" );
        synchronize<%= entity.data_module.name %><%= entity.name %>Removals( dataSourceCode, recorder );
        final int removeCount = recorder.getMetricValue( "<%= entity.data_module.name %>.<%= entity.name %>.RemoveCount" ).get();
        log( dataSourceCode, "Removed " + removeCount + " <%= entity.data_module.name %>.<%= entity.name %> instances" );
      }
<% end -%>
<% data_module.sync.entities_to_synchronize.each do |entity| -%>
      {
        log( dataSourceCode, "Synchronizing <%= entity.data_module.name %>.<%= entity.name %> adds and updates" );
        synchronize<%= entity.data_module.name %><%= entity.name %>Updates( dataSourceCode, recorder );
        final int addCount = recorder.getMetricValue( "<%= entity.data_module.name %>.<%= entity.name %>.AddCount" ).get();
        final int updateCount = recorder.getMetricValue( "<%= entity.data_module.name %>.<%= entity.name %>.UpdateCount" ).get();
        log( dataSourceCode, "Added " + addCount + " <%= entity.data_module.name %>.<%= entity.name %> instances" );
        log( dataSourceCode, "Updated " + updateCount + " <%= entity.data_module.name %>.<%= entity.name %> instances" );
      }
<% end -%>
      log( dataSourceCode, "Successfully synchronized data" );
    }
    catch ( final Throwable e )
    {
      recordSyncError( recorder, SYNC_ERROR_CODE, e );
      log( dataSourceCode, "Synchronisation of data failed", e );
    }
    finally
    {
      completeSynchronization( recorder );
    }
    return toSyncStatus( recorder );
  }

  private void log( final String dataSourceCode, final String message )
  {
    log( dataSourceCode, message, null );
  }

  private void log( final String dataSourceCode, final String message, final Throwable t )
  {
    final String prefix = dataSourceCode + ": ";
    LOG.log( java.util.logging.Level.INFO, prefix + message, t );
  }
<% data_module.sync.entities_to_synchronize.each do |entity| -%>

  @java.lang.SuppressWarnings( "unchecked" )
  @javax.annotation.Nonnull
  private java.util.List<Object[]> query<%= entity.data_module.name %><%= entity.name %>Updates( @javax.annotation.Nonnull final String dataSourceCode )
    throws Exception
  {
    _transaction.begin();
    try
    {
      final String sql = _context.getSqlToRetrieve<%= entity.data_module.name %><%= entity.name %>ListToUpdate( dataSourceCode );
      final javax.persistence.Query query = _entityManager.createNativeQuery( sql );
      query.setParameter( 1, dataSourceCode );
      return query.getResultList();
    }
    finally
    {
      _transaction.commit();
    }
  }

  private void synchronize<%= entity.data_module.name %><%= entity.name %>Updates( @javax.annotation.Nonnull final String dataSourceCode, @javax.annotation.Nonnull final iris.syncrecord.server.service.SynchronizationRecorder recorder )
    throws Exception
  {
    final java.util.List<Object[]> results = query<%= entity.data_module.name %><%= entity.name %>Updates( dataSourceCode );

    for ( final Object[] result : results )
    {
      // Start Replication "Hook"
      _transaction.begin();
      try<commit_msg>Flush to ensure other entities are present in context<commit_after>/* DO NOT EDIT: File is auto-generated */
package <%= to_package(data_module.sync.qualified_sync_ejb_name) %>;

@javax.ejb.Singleton( name = iris.acal.server.service.master.SynchronizationService.NAME )
@javax.ejb.TransactionManagement( javax.ejb.TransactionManagementType.BEAN )
@javax.ejb.ConcurrencyManagement( javax.ejb.ConcurrencyManagementType.BEAN )
public class <%= data_module.sync.sync_ejb_name %>
  extends iris.syncrecord.server.service.StandardSyncService
  implements <%= data_module.service_by_name(:SynchronizationService).ejb.qualified_service_name %>
{
  private static final java.util.logging.Logger LOG = java.util.logging.Logger.getLogger( <%= data_module.sync.sync_ejb_name %>.class.getName() );

  private static final int SYNC_ERROR_CODE = 42;

  @javax.persistence.PersistenceContext( unitName = <%= data_module.repository.jpa.qualified_unit_descriptor_name %>.NAME )
  private javax.persistence.EntityManager _entityManager;
  @javax.annotation.Resource
  private javax.transaction.UserTransaction _transaction;
<% data_module.sync.entities_to_synchronize.each do |entity| -%>
  @javax.inject.Inject
  private <%= entity.dao.jpa.qualified_dao_service_name %> _<%= entity.data_module.name %><%= entity.name %>Repository;
<% end -%>
  @javax.annotation.Resource
  private javax.transaction.TransactionSynchronizationRegistry _registry;
  @javax.inject.Inject
  private org.realityforge.replicant.server.EntityMessageEndpoint _endpoint;
  @javax.inject.Inject
  private <%= data_module.service_by_name(:SynchronizationContext).ejb.qualified_service_name %> _context;

  private volatile boolean _inSync;

  @java.lang.Override
  @javax.annotation.Nonnull
  public iris.syncrecord.server.data_type.SyncStatusDTO synchronizeDataSource( @javax.annotation.Nonnull final String dataSourceCode )
  {
    if ( !_inSync )
    {
      _inSync = true;
      try
      {
        if ( isFeatureEnabled( dataSourceCode + ".syncEnabled" ) )
        {
          return performSynchronization( dataSourceCode );
        }
        else
        {
          return createDisabledSyncStatus( dataSourceCode );
        }
      }
      finally
      {
        _inSync = false;
      }
    }
    else
    {
      return createAlreadyActiveSyncStatus( dataSourceCode );
    }
  }

  private iris.syncrecord.server.data_type.SyncStatusDTO performSynchronization( @javax.annotation.Nonnull final String dataSourceCode )
  {
    final iris.syncrecord.server.service.SynchronizationRecorder recorder = startSynchronization( dataSourceCode );

<% data_module.sync.entities_to_synchronize.each do |entity| -%>
    recorder.incMetric( "<%= entity.data_module.name %>.<%= entity.name %>.AddCount", 0 );
    recorder.incMetric( "<%= entity.data_module.name %>.<%= entity.name %>.RemoveCount", 0 );
    recorder.incMetric( "<%= entity.data_module.name %>.<%= entity.name %>.UpdateCount", 0 );
<% end -%>

    log( dataSourceCode, "Synchronizing data" );
    try
    {
<% data_module.sync.entities_to_synchronize.reverse.each do |entity| -%>
      {
        log( dataSourceCode, "Synchronizing <%= entity.data_module.name %>.<%= entity.name %> removals" );
        synchronize<%= entity.data_module.name %><%= entity.name %>Removals( dataSourceCode, recorder );
        final int removeCount = recorder.getMetricValue( "<%= entity.data_module.name %>.<%= entity.name %>.RemoveCount" ).get();
        log( dataSourceCode, "Removed " + removeCount + " <%= entity.data_module.name %>.<%= entity.name %> instances" );
      }
<% end -%>
<% data_module.sync.entities_to_synchronize.each do |entity| -%>
      {
        log( dataSourceCode, "Synchronizing <%= entity.data_module.name %>.<%= entity.name %> adds and updates" );
        synchronize<%= entity.data_module.name %><%= entity.name %>Updates( dataSourceCode, recorder );
        final int addCount = recorder.getMetricValue( "<%= entity.data_module.name %>.<%= entity.name %>.AddCount" ).get();
        final int updateCount = recorder.getMetricValue( "<%= entity.data_module.name %>.<%= entity.name %>.UpdateCount" ).get();
        log( dataSourceCode, "Added " + addCount + " <%= entity.data_module.name %>.<%= entity.name %> instances" );
        log( dataSourceCode, "Updated " + updateCount + " <%= entity.data_module.name %>.<%= entity.name %> instances" );
      }
<% end -%>
      log( dataSourceCode, "Successfully synchronized data" );
    }
    catch ( final Throwable e )
    {
      recordSyncError( recorder, SYNC_ERROR_CODE, e );
      log( dataSourceCode, "Synchronisation of data failed", e );
    }
    finally
    {
      completeSynchronization( recorder );
    }
    return toSyncStatus( recorder );
  }

  private void log( final String dataSourceCode, final String message )
  {
    log( dataSourceCode, message, null );
  }

  private void log( final String dataSourceCode, final String message, final Throwable t )
  {
    final String prefix = dataSourceCode + ": ";
    LOG.log( java.util.logging.Level.INFO, prefix + message, t );
  }
<% data_module.sync.entities_to_synchronize.each do |entity| -%>

  @java.lang.SuppressWarnings( "unchecked" )
  @javax.annotation.Nonnull
  private java.util.List<Object[]> query<%= entity.data_module.name %><%= entity.name %>Updates( @javax.annotation.Nonnull final String dataSourceCode )
    throws Exception
  {
    _transaction.begin();
    try
    {
      final String sql = _context.getSqlToRetrieve<%= entity.data_module.name %><%= entity.name %>ListToUpdate( dataSourceCode );
      final javax.persistence.Query query = _entityManager.createNativeQuery( sql );
      query.setParameter( 1, dataSourceCode );
      return query.getResultList();
    }
    finally
    {
      _transaction.commit();
    }
  }

  private void synchronize<%= entity.data_module.name %><%= entity.name %>Updates( @javax.annotation.Nonnull final String dataSourceCode, @javax.annotation.Nonnull final iris.syncrecord.server.service.SynchronizationRecorder recorder )
    throws Exception
  {
    final java.util.List<Object[]> results = query<%= entity.data_module.name %><%= entity.name %>Updates( dataSourceCode );

    for ( final Object[] result : results )
    {
      // Start Replication "Hook"
      _transaction.begin();
      _entityManager.flush();
      try
{'fraction_non_alphanumeric': 0.13123609787098825, 'fraction_numerical': 0.0009532888465204957, 'mean_word_length': 3.7848726719878374, 'pattern_counts': {'":': 2, '<': 111, '<?xml version=': 0, '>': 111, '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': '4902607', 'n_tokens_mistral': 3960, 'n_tokens_neox': 3809, 'n_words': 718}